1. def saveDf(df: DataFrame, table: String): Unit = {
    2.   df.createOrReplaceTempView(s"${table}")
    3. }
    6. def makeJsonDf(joinSql: String, jsonColumnName: String, otherCollectFields: String): DataFrame = {
    8.   //先join获取全部字段信息
    10.   //拿到t1和t2两张表
    11.     val karr = joinSql.split(" ")
    12.   val k1 = karr.indexOf("t1")
    13.   val k2 = karr.indexOf("t2")
    15.   var table1 = karr(k1-1)
    16.   var table2 = karr(k2-1)
    18.   val df1: DataFrame = spark.sql("select * from " + table1)
    19.   var df2: DataFrame = spark.sql("select * from " + table2)
    21.   val df: DataFrame = spark.sql(joinSql)
    22.   var realSql: String = joinSql.replace("\n", "").trim()
    23.   println(s"sql: ${realSql}")
    24.   println("join表schema：")
    25.   df1.printSchema()
    26.   println()
    28.   val mfield = realSql.slice("select ".length(), realSql.indexOf(" from "))
    31.   //默认以t1开头的字段作为group by分组字段
    32.   var groupByFields = mfield.split(",").filter(_.contains("t1.")).map(item => {
    33.     val kstr = item.trim()
    34.     if(kstr.contains("distinct ")){
    35.       kstr.substring("distinct t1.".length)
    36.     }
    37.     else{
    38.       kstr.substring(3)
    39.     }
    40.   })
    42.   //不是t1开头的其他字段作为collect字段
    43.   var fields = mfield.split(",").filter(_.contains(".")).filter(!_.contains("t1.")).map(item => {
    44.     val kstr = item.trim()
    45.     if(kstr.contains(" as ")){
    46.       kstr.split(" as ")(1)
    47.     }
    48.     else{
    49.       kstr.substring(3)
    50.     }
    51.   })
    53.   println("group fields....")
    54.   groupByFields.foreach(println)
    56.   if (groupByFields.contains("*")) {
    57.     df1.schema.fieldNames.foreach(one => {
    58.       groupByFields = groupByFields :+ one
    59.     })
    60.   }
    62.   if(fields.contains("*")){
    63.        df2.schema.fieldNames.foreach(one => {
    64.       fields = fields :+ one
    65.     })
    66.   }
    68.   otherCollectFields.split(",").foreach(one => {
    69.     fields = fields :+ one
    70.   })
    72.   var groupByColumns: Array[Column] = Array[Column]()
    73.   groupByFields.distinct.foreach(one => {
    74.     if (df1.columns.contains(one)) {
    75.       groupByColumns = groupByColumns :+ df("t1." + one)
    76.     }
    77.   })
    79.   val mts = df.columns
    80.   mts.foreach(println)
    81.   var collectColumns: Array[Column] = Array[Column]()
    82.   fields.distinct.foreach(field => {
    83.     if (df.columns.contains(field)) {
    84.       collectColumns = collectColumns :+ lit(field)
    85.       collectColumns = collectColumns :+ df("t2." + field)
    86.     }
    87.   })
    89.   println("group fields:")
    90.   groupByColumns.foreach(println)
    91.   println()
    93.   println("collect fields:")
    94.   collectColumns.foreach(println)
    95.   println()
    97.   val df5: DataFrame = df.groupBy(groupByColumns: _*).agg(collect_list(map(collectColumns: _*)).as(jsonColumnName))
    98.   val df6 = df5.withColumn(jsonColumnName, to_json(df5(jsonColumnName)))
    99.   df6
    100. }

    使用示例

    1. var sql = "select t1.registno,t1.accidentno,t1.policyno,t2.casetype as caseflag from ods_new.ods_car_lregist t1 left join ods_new.ods_car_lclaim t2 on t1.accidentno = t2.accidentno and t1.policyno = t2.policyno"
    2. var claimDf = spark.sql(sql)
    3. saveDf(claimDf, "tmp_claim")
    5. // 理赔 join 报案
    6. sql = "select t1.*,t2.frameno,t2.licenseno,t2.reportormobile,t2.comcode,t2.accidentno,t2.policyno from tmp_claim t1 left join  ods_new.ods_car_lregist t2 on t1.accidentno = t2.accidentno and t1.registno = t2.registno";
    7. claimDf = makeJsonDf(sql, "incident", "")
    8. saveDf(claimDf, "tmp_claim")