PySpark explodieren Liste in mehrere Spalten basierend auf Name

Question

Hallo Ich habe es mit einem etwas schwierigen Dateiformat, das ich versuche für einige zukünftige Verarbeitung zu reinigen. Ich habe Pyspark benutzt, um die Daten in einen Datenrahmen zu verarbeiten.

Die Datei sieht wie folgt aus:

AA 1234 ZXYW 
BB A 890 
CC B 321 
AA 1234 LMNO 
BB D 123 
CC E 321 
AA 1234 ZXYW 
CC E 456 

Jeder ‚AA‘ Rekord den Beginn einer logischen Gruppe oder Aufzeichnungen und die Daten in jeder Zeile definiert ist feste Länge und hat darin codierte Information, dass ich möchte extrahieren. Es gibt mindestens 20-30 verschiedene Datensatztypen. Sie werden immer mit einem Zwei-Buchstaben-Code am Anfang jeder Zeile gekennzeichnet. Es können 1 oder viele verschiedene Datensatztypen werden in jeder Gruppe (dh nicht alle Datensatztypen für jede Gruppe vorhanden sind)

Als ersten Schritt habe ich die Aufzeichnungen zusammen in diesem Format Gruppe verwaltet:

+----------------+---------------------------------+ 
|   index|       result| 
+----------------+---------------------------------+ 
|    1|[AA 1234 ZXYV,BB A 890,CC B 321]| 
|    2|[AA 1234 LMNO,BB D 123,CC E 321]| 
|    3|[AA 1234 ZXYV,CC B 321]   | 
+----------------+---------------------------------+ 

Und als zweite Stufe will ich wirklich Daten in die folgenden Spalten in einem Datenrahmen erhalten:

+----------------+---------------------------------+-------------+--------+--------+ 
|   index|       result|   AA|  BB|  CC| 
+----------------+---------------------------------+-------------+--------+--------+ 
|    1|[AA 1234 ZXYV,BB A 890,CC B 321]|AA 1234 ZXYV|BB A 890|CC B 321| 
|    2|[AA 1234 LMNO,BB D 123,CC E 321]|AA 1234 LMNO|BB D 123|CC E 321| 
|    3|[AA 1234 ZXYV,CC B 321]   |AA 1234 ZXYV| Null|CC B 321| 
+----------------+---------------------------------+-------------+--------+--------+ 

an diesem Punkt, weil die Informationen zu extrahieren, die ich sollte trivial sein müssen.

Hat jemand irgendwelche Vorschläge, wie ich das vielleicht könnte?

Vielen Dank.

Answer 1

Sie können dazu flatMap und pivot verwenden. Ausgehend von den Ergebnissen aus der ersten Stufe:

rdd = sc.parallelize([(1,['AA 1234 ZXYV','BB A 890','CC B 321']), 
         (2,['AA 1234 LMNO','BB D 123','CC E 321']), 
         (3,['AA 1234 ZXYV','CC B 321'])]) 

df = rdd.toDF(['index', 'result']) 

Sie können zuerst das Array in mehrere Zeilen explodieren flatMap mit und extrahieren die aus zwei Buchstaben-Kennung in einer separaten Spalte.

df_flattened = df.rdd.flatMap(lambda x: [(x[0],y, y[0:2],y[3::]) for y in x[1]])\ 
       .toDF(['index','result', 'identifier','identifiertype']) 

und verwenden pivot die aus zwei Buchstaben-Kennung in Spaltennamen zu ändern:

df_result = df_flattened.groupby(df_flattened.index,)\ 
         .pivot("identifier")\ 
         .agg(first("identifiertype"))\ 
         .join(df,'index') 

Ich fügte hinzu, die Join

die result Spalt zurück

Answer 2

Alternative Art und Weise Array zu explodieren, ohne zu rdd Umwandlung ,

from pyspark.sql import functions as F 

udf1 = F.udf(lambda x : x.split()[0]) 
df.select('index',F.explode('result').alias('id'),udf1(F.col('id')).alias('idtype')).show() 

+-----+-------------+------+ 
|index|   id|idtype| 
+-----+-------------+------+ 
| 1|AA 1234 ZXYV| AA| 
| 1|  BB A 890| BB| 
| 1|  CC B 321| CC| 
| 2|AA 1234 LMNO| AA| 
| 2|  BB D 123| BB| 
| 2|  CC E 321| CC| 
| 3|AA 1234 ZXYV| AA| 
| 3|  CC B 321| CC| 
+-----+-------------+------+ 

df1.groupby('index').pivot('idtype').agg(F.first('id')).join(df,'index').show() 

Answer 3

Angenommen, Sie sind usi ng Spark 2.x, ich denke, was Sie suchen, ist die Pivot-Operation auf dem Spark-Datenframe.

Zuerst können Sie eine Tabelle mit nur 2 Spalten, die 2-Buchstaben-Codierung und den Rest des Inhalts in einer anderen Spalte erstellen. Dann können Sie Pivot auf dem Datenrahmen verwenden, um dies zu tun, wie unten zu sehen ist.

df.pivot("encoding_col",Seq("AA","BB")) 

Sie können here