Elegante JSON appiattire in Spark

Question

Ho la seguente dataframe scintille:

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

Quello che vorrei fare è appiattirsi questo dataframe in modo che il prop_1 ... prop_n esistono al livello superiore. Cioè

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

Esistono diverse soluzioni per problemi simili. Il meglio che posso trovare è posato here. Tuttavia, la soluzione funziona solo se properties è di tipo Array. Nel mio caso, le proprietà sono di tipo StructType.

Un approccio alternativo sarebbe qualcosa di simile:

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

Ma in questo caso devo specificare esplicitamente ogni riga, e che è poco elegante.

Qual è il modo migliore per risolvere questo problema?

Answer 1

Se non siete alla ricerca di una soluzione ricorsiva poi in 1.6 + dot sintassi con la stella dovrebbe funzionare bene:

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) 

Purtroppo questo non funziona in 1.5 e prima.

In questo caso è possibile estrarre semplicemente le informazioni richieste direttamente dallo schema. Troverai un esempio in Dropping a nested column from Spark DataFrame che dovrebbe essere facile da regolare per adattarsi a questo scenario e un altro (appiattimento dello schema ricorsivo in Python) Pyspark: Map a SchemaRDD into a SchemaRDD.