比较 Pyspark 中的列

我正在开发一个包含 n 列的 PySpark DataFrame。我有一组 m 列 (m

例如：

输入：PySpark DataFrame 包含：

col_1 = [1,2,3], col_2 = [2,1,4], col_3 = [3,2,5]

Ouput :

col_4 = max(col1, col_2, col_3) = [3,2,5]

pandas 中有类似的东西，如中所述this https://stackoverflow.com/questions/27474921/compare-two-columns-using-pandas问题。

在 PySpark 中是否有任何方法可以执行此操作，或者我应该将 PySpark df 转换为 Pandas df 然后执行操作？

您可以减少对列列表使用 SQL 表达式：

from pyspark.sql.functions import max as max_, col, when
from functools import reduce

def row_max(*cols):
    return reduce(
        lambda x, y: when(x > y, x).otherwise(y),
        [col(c) if isinstance(c, str) else c for c in cols]
    )

df = (sc.parallelize([(1, 2, 3), (2, 1, 2), (3, 4, 5)])
    .toDF(["a", "b", "c"]))

df.select(row_max("a", "b", "c").alias("max")))

Spark 1.5+还提供least, greatest

from pyspark.sql.functions import greatest

df.select(greatest("a", "b", "c"))

如果你想保留最大值的名称，你可以使用`structs：

from pyspark.sql.functions import struct, lit

def row_max_with_name(*cols):
    cols_ = [struct(col(c).alias("value"), lit(c).alias("col")) for c in cols]
    return greatest(*cols_).alias("greatest({0})".format(",".join(cols)))

 maxs = df.select(row_max_with_name("a", "b", "c").alias("maxs"))

最后，您可以使用上面的内容来查找选择“顶部”列：

from pyspark.sql.functions import max

((_, c), ) = (maxs
    .groupBy(col("maxs")["col"].alias("col"))
    .count()
    .agg(max(struct(col("count"), col("col"))))
    .first())

df.select(c)