Wie konvertiere ich einen durch Doppelpunkt getrennten String in MapType?

Question

Ich versuche, einen Dataframe in RDD zu konvertieren, um die Karte (mit Schlüssel-Wert-Paar) in eine andere Zeile zu explodieren.

Info = sqlContext.read.format("csv"). \ 
option("delimiter","\t"). \ 
option("header", "True"). \ 
option("inferSchema", "True"). \ 
load("file.tsv") 

DataFrame[ID: int, Date: timestamp, Comments: string] 

Die Beispieldaten in DF sind wie folgt.

ID Date     Comments  
1 2015-04-30 22:42:49.0 {44:'xxxxxxxx'} 
2 2015-05-06 08:53:18.0 {83:'aaaaaaaaa', 175:'bbbbbbbbb', 86:'cccccccccc'} 
3 2015-05-13 19:57:13.0 {487:'yyyyyyyyyyy', 48:'zzzzzzzzzzzzzz'} 

Nun sind die Kommentare bereits in Schlüssel-Wert-Paare, aber es wird als String gelesen, ich jedes Schlüssel-Wert-Paar in einer anderen Zeile zu explodieren möchten. Für z.B.

Expected OUTPUT 
ID Date      Comments 
1 2015-04-30 22:42:49.0 {44:'xxxxxxxx'} 
2 2015-05-06 08:53:18.0 {83:'aaaaaaaaa'} 
2 2015-05-06 08:53:18.0 {175:'bbbbbbbbb'} 
2 2015-05-06 08:53:18.0 {86:'cccccccccc'} 
3 2015-05-13 19:57:13.0 {487:'yyyyyyyyyyy'} 
3 2015-05-13 19:57:13.0 {48:'zzzzzzzzzzzzzz'} 

Ich habe versucht, es zu einem RDD zu konvertieren und flatMap gelten, aber ohne Erfolg. Ich möchte, dass alle Spalten zurückgegeben werden. Ich habe dies versucht:

Info.rdd.flatMap(lambda x: (x['SearchParams'].split(':'), x)) 

Answer 1

Verwenden Sie die mitgelieferten split und explode Funktionen in dem Datenframe-API, die Daten auf „“ zu spalten. Um die Karte zu erstellen, möchten Sie create_map verwenden. Diese Funktion erwartet zwei separate Spalten als Eingabe. Hier unten ist ein Beispiel wurden zwei temporäre Spalten erstellt werden (wieder split verwenden):

Info.withColumn("Comments", explode(split(col("Comments), ", "))) 
    .withColumn("key", split(col("Comments"), ":").getItem(0)) 
    .withColumn("value", split(col("Comments"), ":").getItem(1)) 
    .withColumn("Comments", create_map(col("key"), col("value"))) 

Es sollte möglich sein, diese kürzer zu machen, wie diese (nicht getestet):

Info.withColumn("Comments", split(explode(split(col("Comments), ", ")), ":") 
    .withColumn("Comments", create_map(col("Comments".getItem(0)), col("Comments").getItem(1)))