1

我想根据数据框中现有的列子集创建一个新列（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|
+---+---+---+---+

它由以下人员创建：

rdd1 = sc.parallelize([
                (2,   7.0, 4, 4.0), 
                (99,  2.0, 0, 0.0), 
                (189, 2.4, 0, 0.0)])
d = sqlContext.createDataFrame(rdd1, ('v1', 'v3','v2','v4'))

最终，我想做的是创建另一列 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。任何帮助深表感谢。

4

1 回答 1

3

有数据：

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|
# +---+---+---+---+---+---+
于 2018-02-06T19:37:26.210 回答