Mongoexport在Spark中严格加载json

我现在用这个作为解决方案。但感觉有点粗糙。

case class DataFrameExtended(dataFrame: DataFrame) {

   def undoMongoStrict(): DataFrame = {
    val numberLongType = StructType(List(StructField("$numberLong", StringType, true))) 

    def restructure(fields: Array[StructField], nesting: List[String] = Nil): List[Column] = {
      fields.flatMap(field => {
        val fieldPath = nesting :+ field.name
        val fieldPathStr = fieldPath.mkString(".")
        field.dataType match {
          case dt: StructType if dt == numberLongType =>
            Some(col(s"$fieldPathStr.$$numberLong").cast(LongType).as(field.name))
          case dt: StructType =>
            Some(struct(restructure(dt.fields, fieldPath): _*).as(field.name))
          case _ => Some(col(fieldPathStr).as(field.name))
          //              case dt:ArrayType => //@todo handle other DataTypes Array??
        }
      })
    }.toList


    dataFrame.select(restructure(dataFrame.schema.fields): _*)
  }
}

implicit def dataFrameExtended(df: DataFrame): DataFrameExtended = {
  DataFrameExtended(df)
}

spark
  .read
  .json(inputPath)
  .undoMongoStrict()