Flatmap auf Datenframe

Question

Was ist der beste Weg, um eine flatMap auf einem DataFrame in Funken vorzuformen? Durch Suchen und Testen habe ich zwei verschiedene Ansätze entwickelt. Beide haben einige Nachteile, daher denke ich, dass es einen besseren/einfacheren Weg geben sollte.

Die erste Art, wie ich gefunden habe, ist, zuerst die DataFrame in eine RDD zu konvertieren und dann wieder zurück:

val map = Map("a" -> List("c","d","e"), "b" -> List("f","g","h")) 
val df = List(("a", 1.0), ("b", 2.0)).toDF("x", "y") 

val rdd = df.rdd.flatMap{ row => 
    val x = row.getAs[String]("x") 
    val x = row.getAs[Double]("y") 
    for(v <- map(x)) yield Row(v,y) 
} 
val df2 = spark.createDataFrame(rdd, df.schema) 

Der zweite Ansatz ist ein DataSet zu erstellen, bevor die flatMap (mit den gleichen Variablen wie oben unter Verwendung) und dann zurück zu konvertieren:

val ds = df.as[(String, Double)].flatMap{ 
    case (x, y) => for(v <- map(x)) yield (v,y) 
}.toDF("x", "y") 

Beide Ansätze recht gut funktionieren, wenn die Anzahl der Spalten klein sind, aber ich habe eine Menge von mehr als 2 Spalten haben. Gibt es einen besseren Weg, um dieses Problem zu lösen? Vorzugsweise so, dass keine Konvertierung notwendig ist.

Answer 1

Sie map rdd

val mapDF = Map("a" -> List("c","d","e"), "b" -> List("f","g","h")).toList.toDF("key", "value") 

Ihre dataframe von

Dann erstelle können die join und anwenden explode Funktion

val joinedDF = df.join(mapDF, df("x") === mapDF("key"), "inner") 
    .select("value", "y") 
    .withColumn("value", explode($"value")) 

und Sie erhalten die Lösung

joinedDF.show