動態載入類，構造類的例項，並轉化為DataFrame,同時還要解決入表時，類屬性值與表字段一一對應，同時只有部分欄位有值，其他欄位提供根據欄位型別提供預設值

def mergeRDD(spark:SparkSession,countRDDs:RDD[(String, (Int,Int))],dtoName:String):DataFrame={
  import spark.implicits._
  // 聚合所有指標，並封裝為HashMap ,key 為dto中序號，value為dto 值
val rdd = countRDDs.groupByKey().map(x =>(x._1,x._2.toList)).map{
    x =>
      //從keys分離出相關屬性 ex 1_1_1
 
val keyList = x._1.split("\t").map{y =>
        val fields =y.split("\001")
        val value = fields(2) match {
          case "1" => (fields(1).toInt,fields(0))
          case "2"if(!"null".equals(fields(1))) =>  (fields(1).toString,fields(0))
          case _ =>(null,fields(0))
        }
        value
      }
      keyList++x._2
  }.map{ x =>
    var 
 
map : mutable.HashMap[String,Any] =mutable.HashMap()
    x.map{y =>map.put(y._2.toString,y._1)}
    map
  }
  // 注意此處的scala類在轉df時出現問題，所以採用java類，放棄case class來裝載資料
val classType = Class.forName(dtoName)
  val tx = rdd.map{y =>
    val obj = classType.newInstance()
    val fields = classType.getDeclaredFields
    for 
 
(i <- 1 until fields.length) {
      val field = fields(i)
      val indexKey = i+1+""
field.setAccessible(true)
      if(i==fields.length-1){
        field.set(obj,DateUtil.getSimpleDate(new Date()))
      }else {
        if(field.getType.getName.endsWith("Double")){
          field.set(obj, y.getOrElse(indexKey, 0.00))
        }else if(field.getType.getName.endsWith("Integer")){
          field.set(obj, y.getOrElse(indexKey, 0))
        }else {
          field.set(obj, y.getOrElse(indexKey, ""))
        }
      }
    }
    obj
  }
  //此處的classType是java類
val df = spark.createDataFrame(tx,classType)
  df
}