PySpark：计算行最小值忽略零和空值

Question

我想根据数据框中现有的列子集创建一个新列（v5）。

示例数据框：

+---+---+---+---+
| v1| v2| v3| v4|
+---+---+---+---+
|  2|  4|7.0|4.0|
| 99|  0|2.0|0.0|
|189|  0|2.4|0.0|
+---+---+---+---+

提供示例数据框的另一个视图：

+---+---+---+---+
| v1| v3| v2| v4|
+---+---+---+---+
|  2|7.0|  4|4.0|
| 99|2.0|  0|0.0|
|189|2.4|  0|0.0|
+---+---+---+---+

它的创建者：

+---+---+---+---+
| v1| v2| v3| v4|
+---+---+---+---+
|  2|  4|7.0|4.0|
| 99|  0|2.0|0.0|
|189|  0|2.4|0.0|
+---+---+---+---+

最终，我想做的是创建另一个列 v5，它是与 v1 和 v2 的最小值相对应的值，忽略任一列中存在的零和空值。假设 v1 为键，v3 为值对。同样，v2 为键，v4 为值。例如，在第一行中：在 v1 和 v2 中，最小值属于 v1，即 2，因此 v5 列中的输出应为 7.0 同样，在第二行中：忽略 v1 和 v2 的零值和空值，输出应为成为2.0

原始数据帧有五列作为键，相应的五列作为值所需的输出：

+---+---+---+---+---+
| v1| v2| v3| v4| v5|
+---+---+---+---+---+
|  2|  4|7.0|4.0|7.0|
| 99|  0|2.0|0.0|2.0|
|189|  0|2.4|0.0|2.4|
+---+---+---+---+---+

我试图通过 udf 中的最少函数来完成此操作，但无法使其工作。我正在使用 PySpark 1.6。任何帮助深表感谢。

Answer 1

有数据：

df = spark.createDataFrame([
  (2, 4, 3.0, .0), (99, 0, 2.0, 0.0), (189, 0, 2.4, 0.0)],
  ("v1", "v2", "v3", "v4")
)

您可以将NULL/替换0为-Infor +Inf。

from pyspark.sql.functions import col, lit, least, greatest, when

cols = ["v3", "v4"]

min_ = least(*[
    when(col(c).isNull() | (col(c) == 0), float("inf")).otherwise(col(c))
    for c in cols
]).alias("min")

max_ = greatest(*[
    when(col(c).isNull() | (col(c) == 0), float("-inf")).otherwise(col(c))
    for c in cols
]).alias("max")

并选择：

df.select("*", min_, max_).show()
# +---+---+---+---+---+---+
# | v1| v2| v3| v4|min|max|
# +---+---+---+---+---+---+
# |  2|  4|3.0|7.0|3.0|7.0|
# | 99|  0|2.0|0.0|2.0|2.0|
# |189|  0|2.4|0.0|2.4|2.4|
# +---+---+---+---+---+---+