仅使用Spark-Scala RDD功能将JSON扁平化为表格结构

Question

我已经嵌套了JSON，并且希望以表格结构进行输出。我能够单独解析JSON值，但是在将其制成表格时存在一些问题。我能够通过dataframe轻松实现。但是我想使用“仅RDD”功能。任何帮助，不胜感激。

输入JSON：

  { "level":{"productReference":{  

     "prodID":"1234",

     "unitOfMeasure":"EA"

  },

  "states":[  
     {  
        "state":"SELL",
        "effectiveDateTime":"2015-10-09T00:55:23.6345Z",
        "stockQuantity":{  
           "quantity":1400.0,
           "stockKeepingLevel":"A"
        }
     },
     {  
        "state":"HELD",
        "effectiveDateTime":"2015-10-09T00:55:23.6345Z",
        "stockQuantity":{  
           "quantity":800.0,
           "stockKeepingLevel":"B"
        }
     }
  ] }}

预期产量：

我尝试了以下Spark代码。但是获取这样的输出和Row（）对象无法解析此内容。

079562193，EA，列表（SELLABLE，HELD），列表（2015-10-09T00：55：23.6345Z，2015-10-09T00：55：23.6345Z），列表（1400.0，800.0），列表（SINGLE，SINGLE）

def main(Args : Array[String]): Unit = {

  val conf = new SparkConf().setAppName("JSON Read and Write using Spark RDD").setMaster("local[1]")
  val sc = new SparkContext(conf)
  val sqlContext = new SQLContext(sc)

  val salesSchema = StructType(Array(
    StructField("prodID", StringType, true),
    StructField("unitOfMeasure", StringType, true),
    StructField("state", StringType, true),
    StructField("effectiveDateTime", StringType, true),
    StructField("quantity", StringType, true),
    StructField("stockKeepingLevel", StringType, true)
  ))

  val ReadAlljsonMessageInFile_RDD = sc.textFile("product_rdd.json")

  val x = ReadAlljsonMessageInFile_RDD.map(eachJsonMessages => {

        parse(eachJsonMessages)

      }).map(insideEachJson=>{
        implicit  val formats = org.json4s.DefaultFormats

       val prodID = (insideEachJson\ "level" \"productReference" \"TPNB").extract[String].toString
       val unitOfMeasure = (insideEachJson\ "level" \ "productReference" \"unitOfMeasure").extract[String].toString

       val state= (insideEachJson \ "level" \"states").extract[List[JValue]].
          map(x=>(x\"state").extract[String]).toString()
       val effectiveDateTime= (insideEachJson \ "level" \"states").extract[List[JValue]].
         map(x=>(x\"effectiveDateTime").extract[String]).toString
      val quantity= (insideEachJson \ "level" \"states").extract[List[JValue]].
         map(x=>(x\"stockQuantity").extract[JValue]).map(x=>(x\"quantity").extract[Double]).
         toString
      val stockKeepingLevel= (insideEachJson \ "level" \"states").extract[List[JValue]].
         map(x=>(x\"stockQuantity").extract[JValue]).map(x=>(x\"stockKeepingLevel").extract[String]).
       toString

      //Row(prodID,unitOfMeasure,state,effectiveDateTime,quantity,stockKeepingLevel)

    println(prodID,unitOfMeasure,state,effectiveDateTime,quantity,stockKeepingLevel)

      }).collect()

    //  sqlContext.createDataFrame(x,salesSchema).show(truncate = false)

}

Answer 1

下面的HI是我开发的“仅数据帧”解决方案。寻找完整的“仅RDD”解决方案

def main（Args：Array [String]）：Unit = {

    val conf = new SparkConf（）。setAppName（“使用Spark DataFrame进行JSON读取和写入还有更多选项”）.setMaster（“ local [1]”）
    val sc = new SparkContext（conf）
    val sqlContext =新的SQLContext（sc）

    val sourceJsonDF = sqlContext.read.json（“ product.json”）

         val jsonFlatDF_level = sourceJsonDF.withColumn（“ explode_states”，explode（$“ level.states”））
        .withColumn（“ explode_link”，explode（$“ level._link”））
      .select（$“ level.productReference.TPNB” .as（“ TPNB”），
        $“ level.productReference.unitOfMeasure” .as（“ level_unitOfMeasure”），
        $“ level.locationReference.location” .as（“ level_location”），
        $“ level.locationReference.type” .as（“ level_type”），
        $“ explode_states.state” .as（“ level_state”），
        $“ explode_states.effectiveDateTime” .as（“ level_effectiveDateTime”），
        $“ explode_states.stockQuantity.quantity” .as（“ level_quantity”），
        $“ explode_states.stockQuantity.stockKeepingLevel” .as（“ level_stockKeepingLevel”），
        $“ explode_link.rel” .as（“ level_rel”），
        $“ explode_link.href” .as（“ level_href”），
        $“ explode_link.method” .as（“ level_method”））
jsonFlatDF_oldLevel.show（）

  }