让我以一个简单的例子来解释我要做的事情。让我们说我们有两个非常简单的数据框:
Df1
+---+---+---+
| a1| a2| a3|
+---+---+---+
| 2| 3| 7|
| 1| 9| 6|
+---+---+---+
Df2
+---+---+
| b1| b2|
+---+---+
| 10| 2|
| 9| 3|
+---+---+
来自DF1,DF2,我们需要创建一个新的DF,其中包含DF1,DF2原始列的笛卡尔产品的列。特别是,新的DF将具有" A1B1"," A1B2"," A2B1"," A2B2"," A3B1"," A3B2",而行将是DF1,DF2的相应列的乘法。结果DF应该看起来如下:
Df3
+----+----+----+----+----+----+
|a1b1|a1b2|a2b1|a2b2|a3b1|a3b2|
+----+----+----+----+----+----+
| 20| 4| 30| 6| 70| 14|
| 9| 3| 81| 27| 54| 18|
+----+----+----+----+----+----+
我已经搜索了Spark Online文档以及这里发布的问题,但似乎它们都是关于排的笛卡尔产品,而不是列。例如,rdd.cartesian()提供了笛卡尔产品,这些产品具有不同的值组合,例如以下代码:
r = sc.parallelize([1, 2])
r.cartesian(r).toDF().show()
+---+---+
| _1| _2|
+---+---+
| 1| 1|
| 1| 2|
| 2| 1|
| 2| 2|
+---+---+
,但这不是我需要的。同样,我需要创建新列而不是行。我的问题中的行数将保持不变。我知道UDF最终可以解决问题。但是,在我的真实应用程序中,我们有巨大的数据集,它花费了太长时间来创建所有列(大约500列作为列的所有可能组合)。我们更喜欢具有某种可能提高效率的向量操作。我可能错了,但是Spark UDF似乎是基于行操作,这可能是完成这么长时间才能完成的原因。
非常感谢您的任何建议/反馈/评论。
为了方便起见,我在此处附加了简单的代码,以创建上面显示的示例数据框:
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()
据我所知,它并不简单。这是使用评估:
的照片# 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中的ElementWise产品,但这也同样复杂。您将元素从一个列表中获取多个元素明智的列表到另一个列表,然后将功能向量扩展回数据框。