Konvertieren Pyspark Dataframe Spalte von Array zu neuen Spalten

Question

ich eine Pyspark Datenrahmen mit dieser Struktur haben:

root 
|-- Id: string (nullable = true) 
|-- Q: array (nullable = true) 
| |-- element: struct (containsNull = true) 
| | |-- pr: string (nullable = true) 
| | |-- qt: double (nullable = true) 

Etwas ähnliches wie:

+----+--------------------- ... --+ 
| Id |   Q    | 
+----+---------------------- ... -+ 
| 001| [ [pr1,1.9], [pr3,2.0]...] | 
| 002| [ [pr2,1.0], [pr9,3.9]...] | 
| 003| [ [pr2,9.0], ...   ] | 
    ... 

Ich mag Wold Q Array in Spalten konvertieren (Name pr Wert qt). Auch möchte ich doppelte Spalten durch Zusammenführen (Hinzufügen) der gleichen Spalten vermeiden.

+----+-----+-----+------+ ... ----+ 
| Id | pr1 | pr2 | pr3 | ... prn | 
+----+-----+-----+------+ ... ----+ 
| 001| 1.9 | 0.0 | 2.0 | ...  | 
| 002| 0.0 | 1.0 | 0 | ...  | 
| 003| 0.0 | 9.0 | ... | ...  | 
    ... 

Wie kann ich diese Transformation durchführen ?. Thakyou im Voraus !!. Julián.

Answer 1

Sie können dies mit einer Kombination aus explode und pivot:

import pyspark.sql.functions as F 

# explode to get "long" format 
df=df.withColumn('exploded', F.explode('Q')) 

# get the name and the name in separate columns 
df=df.withColumn('name', F.col('exploded').getItem(0)) 
df=df.withColumn('value', F.col('exploded').getItem(1)) 

# now pivot 
df.groupby('Id').pivot('name').agg(F.max('value')).na.fill(0) 

Answer 2

Sehr interessante Frage. So habe ich es angegangen.

test.csv

001,pr1:0.9,pr3:1.2,pr2:2.0 
002,pr3:5.2,pr4:0.99 

Pyspark

file = sc.textFile("file:///test2.csv") 

//get it in (key,value) 
//[(u'001', u'pr1:0.9')...] 

//rdd1 = file.map(lambda r: r.replace(",","\t",1)).map(lambda r: r.split("\t")).map(lambda r: (r[0],r[1])).flatMapValues(lambda r: r.split(',')) 
rdd1 = file.map(lambda r: r.split(",")[0]).map(lambda r: (r[0],r[1])).flatMapValues(lambda r: r.split(',')) 

//create a DF with 3 columns 
//[(u'001', u'pr1', u'0.9')...)] 
+---+---+----+ 
| _1| _2| _3| 
+---+---+----+ 
|001|pr1| 0.9| 
|001|pr3| 1.2| 
|001|pr2| 2.0| 
|002|pr3| 5.2| 
|002|pr4|0.99| 
+---+---+----+ 


rdd2 = rdd1.map(lambda r: (r[0],r[1].split(":"))).map(lambda r: (r[0],r[1][0],r[1][1])) 
df = rdd2.toDF() 


//Perform the magic 
df.groupBy("_1").pivot("_2").agg(expr("coalesce(first(_3),0)")) 


+---+---+---+---+----+ 
| _1|pr1|pr2|pr3| pr4| 
+---+---+---+---+----+ 
|001|0.9|2.0|1.2| 0| 
|002| 0| 0|5.2|0.99| 
+---+---+---+---+----+