优雅的Json在Spark中展平

Question

我在spark中有以下数据框:

val test = sqlContext.read.json(path = "/path/to/jsonfiles/*")  
test.printSchema
root
 |-- properties: struct (nullable = true)
 |    |-- prop_1: string (nullable = true)
 |    |-- prop_2: string (nullable = true)
 |    |-- prop_3: boolean (nullable = true)
 |    |-- prop_4: long (nullable = true)
...

我想做的是压扁这个数据帧,使其prop_1 ... prop_n存在于顶层.即

test.printSchema
root
|-- prop_1: string (nullable = true)
|-- prop_2: string (nullable = true)
|-- prop_3: boolean (nullable = true)
|-- prop_4: long (nullable = true)
...

有几种类似问题的解决方案.我能找到的最好的就是这里.但是,只有properties类型的解决方案才有效Array.在我的例子中,属性是类型StructType.

另一种方法是:

test.registerTempTable("test")
val test2 = sqlContext.sql("""SELECT properties.prop_1, ... FROM test""")

但在这种情况下,我必须明确指定每一行,这是不优雅的.

解决这个问题的最佳方法是什么？

Answer 1

如果你不是在寻找一个递归解决方案,那么使用star的1.6+点语法应该可以正常工作:

val df = sqlContext.read.json(sc.parallelize(Seq(
  """{"properties": {
       "prop1": "foo", "prop2": "bar", "prop3": true, "prop4": 1}}"""
)))

df.select($"properties.*").printSchema
// root
//  |-- prop1: string (nullable = true)
//  |-- prop2: string (nullable = true)
//  |-- prop3: boolean (nullable = true)
//  |-- prop4: long (nullable = true)

不幸的是,这在1.5及之前不起作用.

在这种情况下,您可以直接从架构中提取所需信息.您可以在Spark DataFrame中删除嵌套列中找到一个示例,该列应该很容易调整以适应此场景,另一个示例(Python中的递归模式展平)Pyspark:将SchemaRDD映射到SchemaRDD.