Mongoexport strict json load in Spark

Question

私はmongoexportを使ってmongodbからデータをエクスポートするプロセスを持っています。

：私のScalaのケースクラスは次のように定義されるよう

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

：documentationとして は、すべてのJSON出力は厳格モードで

である。これは、データが次のようになります意味言及します

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

データの読み込みはもちろん失敗します：

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

strictモードなしモンゴからの輸出のいずれかにするか、手動で適切な構造に各フィールドを再構築することなく、スカラ座でJSONをインポートする方法はありますか？

Answer 1

これを解決策として使用しています。それはややハッキリと感じます。

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