Spark高级操作之json复杂和嵌套数据结构的操作二

原数据：
1593136280858|{“cm”:{“ln”:"-55.0",“sv”:“V2.9.6”,“os”:“8.0.4”,“g”:“C6816QZ0@gmail.com”,“mid”:“489”,“nw”:“3G”,“l”:“es”,“vc”:“4”,“hw”:“640960",“ar”:“MX”,“uid”:“489”,“t”:“1593123253541”,“la”:“5.2”,“md”:“sumsung-18”,“vn”:“1.3.4”,“ba”:“Sumsung”,“sr”:“I”},“ap”:“app”,“et”:[{“ett”:“1593050051366”,“en”:“loading”,“kv”:{“extend2”:"",“loading_time”:“14”,“action”:“3”,“extend1”:"",“type”:“2”,“type1”:“201”,“loading_way”:“1”}},{“ett”:“1593108791764”,“en”:“ad”,“kv”:{“activityId”:“1”,“displayMills”:“78522”,“entry”:“1”,“action”:“1”,“contentType”:“0”}},{“ett”:“1593111271266”,“en”:“notification”,“kv”:{“ap_time”:“1593097087883”,“action”:“1”,“type”:“1”,“content”:""}},{“ett”:“1593066033562”,“en”:“active_background”,“kv”:{“active_source”:“3”}},{“ett”:“1593135644347”,“en”:“comment”,“kv”:{“p_comment_id”:1,“addtime”:“1593097573725”,“praise_count”:973,“other_id”:5,“comment_id”:9,“reply_count”:40,“userid”:7,“content”:“辑赤蹲慰鸽抿肘捎”}}]}
1593136280858|{“cm”:{“ln”:"-114.9",“sv”:“V2.7.8”,“os”:“8.0.4”,“g”:“NW0S962J@gmail.com”,“mid”:“490”,“nw”:“3G”,“l”:“pt”,“vc”:“8”,“hw”:"6401136”,“ar”:“MX”,“uid”:“490”,“t”:“1593121224789”,“la”:"-44.4",“md”:“Huawei-8”,“vn”:“1.0.1”,“ba”:“Huawei”,“sr”:“O”},“ap”:“app”,“et”:[{“ett”:“1593063223807”,“en”:“loading”,“kv”:{“extend2”:"",“loading_time”:“0”,“action”:“3”,“extend1”:"",“type”:“1”,“type1”:“102”,“loading_way”:“1”}},{“ett”:“1593095105466”,“en”:“ad”,“kv”:{“activityId”:“1”,“displayMills”:“1966”,“entry”:“3”,“action”:“2”,“contentType”:“0”}},{“ett”:“1593051718208”,“en”:“notification”,“kv”:{“ap_time”:“1593095336265”,“action”:“2”,“type”:“3”,“content”:""}},{“ett”:“1593100021275”,“en”:“comment”,“kv”:{“p_comment_id”:4,“addtime”:“1593098946009”,“praise_count”:220,“other_id”:4,“comment_id”:9,“reply_count”:151,“userid”:4,“content”:“抄应螟皮釉倔掉汉蛋蕾街羡晶”}},{“ett”:“1593105344120”,“en”:“praise”,“kv”:{“target_id”:9,“id”:7,“type”:1,“add_time”:“1593098545976”,“userid”:8}}]}

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataframe, SparkSession}


object b {
  def main(args: Array[String]): Unit = {

    val sparkSession: SparkSession = SparkSession.builder().appName("b").master("local[*]").getOrCreate()
    val sc: SparkContext = sparkSession.sparkContext

    import sparkSession.implicits._
    import org.apache.spark.sql.types._
    import org.apache.spark.sql.functions._

    //装载JSON数据源
    val optionRDD: RDD[String] = sc.textFile("in/op.log")
//    optionARR.foreach(println)

    //拆分数据源
    val option1: RDD[(String, String)] = optionRDD.map(x=>x.split('|')).map(x=>(x(0),x(1)))
//    option1.foreach(println)

//    val jsonStr: RDD[String] = option1.map(x=>x._2)
//    jsonStr.foreach(println)


    val jsonStr: RDD[String] = option1.map(
      x => {
        val jsstr = x._2;  //数据
        val js2 = jsstr.substring(0, jsstr.length - 1);
        js2 + ", "id":" + x._1 + " }" //数据+id
      })

    val jsonstrDF: Dataframe = jsonStr.toDF()

    jsonstrDF.show(false)
//    jsonstrDF.printSchema()

    val jsonobj: Dataframe = jsonstrDF.select(
      get_json_object($"value", "$.cm").as("cm"),
      get_json_object($"value", "$.ap").as("ap"),
      get_json_object($"value", "$.et").as("et")
    )
    jsonobj.show(false)
//    jsonobj.printSchema()

    val jsonobj2: Dataframe = jsonobj.select($"ap",
      get_json_object($"cm", "$.ln").as("ln"),
      get_json_object($"cm", "$.sv").as("sv"),
      get_json_object($"cm", "$.os").as("os"),
      get_json_object($"cm", "$.g").as("g"),
      get_json_object($"cm", "$.mid").as("mid"),
      get_json_object($"cm", "$.nw").as("nw"),
      get_json_object($"cm", "$.l").as("l"),
      get_json_object($"cm", "$.vc").as("vc"),
      get_json_object($"cm", "$.hw").as("hw"),
      get_json_object($"cm", "$.ar").as("ar"),
      get_json_object($"cm", "$.uid").as("uid"),
      get_json_object($"cm", "$.t").as("t"),
      get_json_object($"cm", "$.la").as("la"),
      get_json_object($"cm", "$.md").as("md"),
      get_json_object($"cm", "$.vn").as("vn"),
      get_json_object($"cm", "$.ba").as("ba"),
      get_json_object($"cm", "$.sr").as("sr"),
      $"et")

    jsonobj2.show(false)
//    jsonobj2.schema

      val jsonStr2Df: Dataframe = jsonobj2.select($"et")
    jsonStr2Df.show(false)
//    jsonStr2Df.schema

    val schema: ArrayType = ArrayType(
      StructType(
        StructField("ett", StringType) :: StructField("en", StringType) :: StructField("kv", StringType) :: Nil
      )
    )
//    println(schema)

    val jsonobj3: Dataframe = jsonStr2Df.select(from_json($"et",schema).as("events"))

    jsonobj3.show(false)
//    jsonobj3.schema

    val jsonobj4: Dataframe = jsonobj3.withColumn("events",explode(col("events")))
      jsonobj4.show(false)
//      jsonobj4.schema

    jsonobj4.select($"events.ett",$"events.en",$"events.kv").show(false)

    jsonobj2.select(
      $"ap",$"ln",$"sv",$"os",
      $"g",$"mid",$"nw",$"l",
      $"vc",$"hw",$"ar",$"uid",
      $"t",$"la",$"md",$"vn",
      $"ba",$"sr",from_json($"et",schema).as("events")
    ).withColumn("events", explode(col("events"))).select($"ap",$"ln",$"sv",$"os",
      $"g",$"mid",$"nw",$"l",
      $"vc",$"hw",$"ar",$"uid",
      $"t",$"la",$"md",$"vn",
      $"ba",$"sr",$"events.ett", $"events.en", $"events.kv").show(false)

  }
}