Wie man neue Spalten basierend auf kartesischen Produkt von mehreren Spalten von pyspark Datenrahmen erstellen

Question

Lassen Sie mich ein einfaches Beispiel zu erklären, was ich versuche zu tun. lassen Sie uns sagen, dass wir, wie unten zwei sehr einfache Datenrahmen haben:

Df1 
+---+---+---+ 
| a1| a2| a3| 
+---+---+---+ 
| 2| 3| 7| 
| 1| 9| 6| 
+---+---+---+ 

Df2 
+---+---+ 
| b1| b2| 
+---+---+ 
| 10| 2| 
| 9| 3| 
+---+---+ 

Von DF1, DF2, brauchen wir eine neue df mit Spalten erstellen, die cartesianischen Produkt von Original-Spalten von DF1, DF2 sind. Insbesondere wird das neue df "a1b1", "a1b2", "a2b1", "a2b2", "a3b1", "a3b2" haben, und die Zeilen werden die Multiplikation der entsprechenden Spalten von df1, df2 sein. Ergebnis df sollte wie folgt aussehen:

Df3 
+----+----+----+----+----+----+ 
|a1b1|a1b2|a2b1|a2b2|a3b1|a3b2| 
+----+----+----+----+----+----+ 
| 20| 4| 30| 6| 70| 14| 
| 9| 3| 81| 27| 54| 18| 
+----+----+----+----+----+----+ 

ich gesucht habe Funken Online-Dokumentation sowie Fragen hier gepostet, aber es scheint, dass sie alle über kartesisches Produkt von Zeilen, Spalten nicht. Zum Beispiel rdd.cartesian() liefert kartesisches Produkt unterschiedlicher Kombination von Werten in der Reihe, wie der folgenden Code:

r = sc.parallelize([1, 2]) 
r.cartesian(r).toDF().show() 

+---+---+ 
| _1| _2| 
+---+---+ 
| 1| 1| 
| 1| 2| 
| 2| 1| 
| 2| 2| 
+---+---+ 

Aber das ist nicht das, was ich brauche. Auch hier muss ich neue Spalten anstelle von Zeilen erstellen. Die Anzahl der Zeilen bleibt in meinem Problem gleich. Ich verstehe, udf kann das Problem schließlich lösen. In meiner realen Anwendung haben wir jedoch einen riesigen Datensatz, der zu lange dauert, um alle Spalten zu erstellen (etwa 500 neue Spalten als die möglichen Kombinationen von Spalten). Wir bevorzugen einige Arten von Vektoroperationen, die die Effizienz erhöhen können. Ich mag mich irren, aber Spark UDF scheint auf Reihenoperationen zu basieren, was der Grund sein könnte, warum es so lange gedauert hat.

Vielen Dank für Anregungen/Feedback/Kommentare.

Für Ihre Bequemlichkeit angebracht ich den einfachen Code hier das Beispiel erstellen Datenrahmen oben gezeigt:

df1 = sqlContext.createDataFrame([[2,3,7],[1,9,6]],['a1','a2','a3']) 
df1.show() 

df2 = sqlContext.createDataFrame([[10,2],[9,3]],['b1','b2']) 
df2.show() 

Answer 1

Es ist nicht einfach, wie weit ich weiß. Hier ist ein Schuss auf ihn eval:

# function to add rownumbers in a dataframe 
def addrownum(df): 
    dff = df.rdd.zipWithIndex().toDF(['features','rownum']) 
    odf = dff.map(lambda x : tuple(x.features)+tuple([x.rownum])).toDF(df.columns+['rownum']) 
    return odf 

df1_ = addrownum(df1) 
df2_ = addrownum(df2) 
# Join based on rownumbers 
outputdf = df1_.rownum.join(df2_,df1_.rownum==df2_.rownum).drop(df1_.rownum).drop(df2_.rownum) 

n1 = ['a1','a2','a3'] # columns in set1 
n2 = ['b1','b2']  # columns in set2 

# I create a string of expression that I want to execute 
eval_list = ['x.'+l1+'*'+'x.'+l2 for l1 in n1 for l2 in n2] 
eval_str = '('+','.join(eval_list)+')' 
col_list = [l1+l2 for l1 in n1 for l2 in n2] 

dfcartesian = outputdf.map(lambda x:eval(eval_str)).toDF(col_list) 

Etwas anderes, die hilfreich sein könnten, um Sie element Produkt in spark.ml.feature ist, aber es wird nicht weniger komplex sein. Sie nehmen Elemente von einer Liste, die mehrere Elemente enthält, zu der anderen Liste und erweitern die Merkmalsvektoren zurück zu einem Datenrahmen.