在PySpark中为基于行的操作添加列

Question

我有一个Spark DataFrame，我正在尝试基于以前的列创建一个新列，但对我来说，困难的是我已经按行计算了列的值。例如：

列1 |列2 |列3

1 | 2 | 3

4|5|0

3 | 1 | 1

因此，我想要一个新列，每行具有表达式
max（col1， col2， col3）的列名。所以，所需的输出：

Col1 |Col2 |Col3 |Col4

1|2|3|'col3'

4|5|0|'col2'

3|1|1|'col1'

总之在PySpark是可能做到的？

Answer 1

这不是一个理想的答案，因为它迫使你回到RDD。如果我找到了一个更好的方法，让你留在DataFrame宇宙中，我会更新我的答案。但目前这应该奏效。

a = sc.parallelize([[1,2,3],[4,5,0],[3,1,1]])
headers = ["col1", "col2", "col3"]

b = a.map(lambda x: (x[0], x[1], x[2], headers[x.index(max(x))]))

b.toDF(headers.append("max_col")).show()

这基本上允许您通过逐行迭代RDD来使用python中的< code>max操作。然后，它通过索引标题列表找到正确的列。

再次，我不确定这是最好的方法，我希望找到更好的方法。