Mongoexport strenge JSON Last in Spark

Question

Ich habe einen Prozess, der Daten von einem Mongodb mit mongoexport exportiert. Als documentation erwähnt alle json Ausgabe ist in Strict Modus

Dies bedeutet, Daten wie folgt aussehen:

"{amount":{"$numberLong":"3"},"count":{"$numberLong":"245"}} 

Wo, wie meine Scala Fallklasse definiert ist als:

case class MongoData(amount: Long, count: Long) 

Lesen der Daten wird natürlich wie folgt fehlschlagen:

spark 
     .read 
     .json(inputPath) 
     .as[MongoData] 

Gibt es eine Möglichkeit, entweder den Export von Mongo ohne den strikten Modus oder die json in Scala zu importieren, ohne jedes Feld auf die entsprechende Struktur manuell Umstrukturierung?

Answer 1

Ich verwende dies jetzt als Lösung. aber es fühlt sich etwas hacky an.

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()