spark-dataframe - 如何读取带有没有值的字段的JSON作为Spark DataFrame？

Question

我正在使用spark-dataframe上的Json文件。我正在尝试使用以下Json字符串解析文件:

{“id”:“00010005”，“time_value”:864359000，“speed”:1079，“acceleration”:19，“la”:36.1433530，“lo”:-11.51577690}
{“id”:“00010005”，“time_value”:864360000，“speed”:1176，“acceleration”:10，“la”:36.1432660，“lo”:-11.51578220}
{“id”:“00010005”，“time_value”:864361000，“speed”:1175，“acceleration”:，“la”:36.1431730，“lo”:-11.51578840}
{“id”:“00010005”，“time_value”:864362000，“speed”:1174，“acceleration”:，“la”:36.1430780，“lo”:-11.51579410}
{“id”:“00010005”，“time_value”:864363000，“speed”:1285，“acceleration”:11，“la”:36.1429890，“lo”:-11.51580110}

这里的加速度字段有时不包含任何值.Spark将那些没有加速度值的json标记为Corrupt_record。

val df = sqlContext.read.json(data)
scala> df.show(20)
+--------------------+------------+--------+---------+-----------+-----+----------+
|     _corrupt_record|acceleration|      id|       la|         lo|speed|time_value|
+--------------------+------------+--------+---------+-----------+-----+----------+
|                null|          -1|00010005|36.143418|-11.5157712|  887| 864358000|
|                null|          19|00010005|36.143353|-11.5157769| 1079| 864359000|
|                null|          10|00010005|36.143266|-11.5157822| 1176| 864360000|
|{"id":"00010005",...|        null|    null|     null|       null| null|      null|
|{"id":"00010005",...|        null|    null|     null|       null| null|      null|

我不想删除这些记录。读取这些Json记录的正确方法是什么？

我尝试了下面的代码，并用“0”值替换了“加速度”。但是它不是解决任何字段的值可能丢失的情况的通用解决方案。

val df1 = df.select("_corrupt_record").na.drop()
val stripRdd = df1.rdd.map( x => x.getString(0)).map(x=>x.replace(""""acceleration":""",""""acceleration":0"""))
val newDf = sqlContext.read.json(stripRdd)
val trimDf = df.drop("_corrupt_record").na.drop
val finalDf = trimDf.unionAll(newDf)

Answer 1

如果您有适合记录的架构，则可以轻松地做到这一点，比如说该架构称为SpeedRecord，其中包含以下字段:加速度，id，la，lo，速度，time_value

case class SpeedRecord(acceleration : Int, id : Long, la : Double , lo : Double, speed : Int, time_value : Long)

val schema = Encoders.bean(classOf[SpeedRecord]).schema
val speedRecord = spark.read.schema(schema).json("/path/data.json")
speedRecord.show()