Как создать новые столбцы на основе декартовых произведений из нескольких столбцов из фреймов данных pyspark

Question

Позвольте мне привести простой пример, чтобы объяснить, что я пытаюсь сделать. давайте говорить, что мы имеем два очень простых dataframes, как показано ниже:

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

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

От df1, df2, нам нужно создать новый ДФ с колоннами, которые декартово произведение исходных столбцов из df1, df2. В частности, новый df будет иметь «a1b1», «a1b2», «a2b1», «a2b2», «a3b1», «a3b2», а строки будут умножением соответствующих столбцов из df1, df2. Результат ДФ должен выглядеть следующим образом:

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

Я искал искровые онлайн документы, а также вопросы размещены здесь, но это, кажется, что они все о декартово произведение строк, а не столбцов. Например, rdd.cartesian() обеспечивает декартово произведение различных комбинаций значений в строке, как следующий код:

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

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

Но это не то, что мне нужно. Опять же, мне нужно создать новые столбцы вместо строк. Количество строк останется таким же в моей проблеме. Я понимаю, что udf может в конечном итоге решить проблему. Однако в моем реальном приложении у нас есть огромный набор данных, который занимает слишком много времени, чтобы создать все столбцы (около 500 новых столбцов - все возможные комбинации столбцов). мы предпочитаем иметь какие-то векторные операции, которые могут повысить эффективность. Возможно, я ошибаюсь, но искривление udf похоже на то, что оно основано на операциях с строками, что может быть причиной, по которой так долго заканчивалось.

Большое спасибо за любые предложения/отзывы/комментарии.

Для вашего удобства я прикрепил простой код здесь, чтобы создать пример dataframes показано выше:

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

Его не так просто, насколько я знаю. Вот снимок на него с помощью 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) 

Что-то другое, что может оказаться полезным для вас поэлементно продукта в spark.ml.feature, но это будет не менее сложным. Вы берете элементы из одного списка из нескольких элементов в другой список и расширяете векторы объектов обратно в dataframe.