withColumn 不允许我使用 max() 函数生成新列

我有一个这样的数据集：

a = sc.parallelize([[1,2,3],[0,2,1],[9,8,7]]).toDF(["one", "two", "three"])

我想要一个数据集添加一个新列，该新列等于其他三列中的最大值。 输出如下所示：

+----+----+-----+-------+
|one |two |three|max_col|
+----+----+-----+-------+
|   1|   2|    3|      3|
|   0|   2|    1|      2|
|   9|   8|    7|      9|
+----+----+-----+-------+

我以为我会用withColumn，像这样：

b = a.withColumn("max_col", max(a["one"], a["two"], a["three"]))

但这会产生错误

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/opt/spark152/python/pyspark/sql/column.py", line 418, in __nonzero__
    raise ValueError("Cannot convert column into bool: please use '&' for 'and', '|' for 'or', "
ValueError: Cannot convert column into bool: please use '&' for 'and', '|' for 'or', '~' for 'not' when building DataFrame boolean expressions.

奇怪的。做max返回一个bool？不根据的文档max https://docs.python.org/2/library/functions.html#max。好的。诡异的。

我觉得这很奇怪：

b = a.withColumn("max_col", a["one"] + a["two"] + a["three"]))

它有效的事实让我更加强烈地认为max的行为方式我不明白。

我也尝试过b = a.withColumn("max_col", max([a["one"], a["two"], a["three"]]))，它将三列作为列表而不是 3 个单独的元素传递。这会产生与上面相同的错误。

实际上你需要的是greatest not max:

from pyspark.sql.functions import greatest

a.withColumn("max_col", greatest(a["one"], a["two"], a["three"]))

为了完整起见，您可以使用least找到最小值：

from pyspark.sql.functions import least

a.withColumn("min_col", least(a["one"], a["two"], a["three"]))

关于您看到的错误非常简单。max取决于丰富的比较。当您比较两列时，您会得到Column:

type(col("a") < col("b")
## pyspark.sql.column.Column

PySpark 明确禁止将列转换为布尔值（您可以检查Column.__nonzero__来源）因为它根本毫无意义。它只是一个逻辑表达式，无法在驱动程序上下文中进行计算。