组聚合中的算术减法 PySpark答案

【问题标题】：Arithmetic subtraction in group aggregation PySpark组聚合中的算术减法 PySpark
【发布时间】：2021-07-03 19:26:42
【问题描述】：

我有以下数据框：

ID val1 val2 val3 ...
1   4    1    3   ...
1   5    4    8   ...
2   6    3    6   ...
2   9    2    2   ...
3   2    1    4   ...
3   1    1    4   ...

我需要按 ID 分组/聚合并减去值，产生以下输出：

ID val1 val2 val3 ...
1   -1   -3  -5   ... 
2   -3    1   4   ...
3    1    0   0   ...

我目前的方法会一次产生 1 列所需的输出：

from pyspark.sql.functions import first, last
output = df.groupBy('id').agg(first('val1') - (last(col('val1'))))

但是，我的数据集有很多列，我需要为所有列找到一种干净的方法。

【问题讨论】：

标签： python apache-spark pyspark

【解决方案1】：

检查下面的代码。

df
.groupBy(col("id"))
.agg(
    (first(col("val1")) - last(col("val1"))).as("val1"),
    (first(col("val2")) - last(col("val2"))).as("val2"),
    (first(col("val3")) - last(col("val3"))).as("val3")
)
.orderBy(col("id"), ascending=True)
.show(false)

+---+----+----+----+
|id |val1|val2|val3|
+---+----+----+----+
|1  |-1  |-3  |-5  |
|2  |-3  |1   |4   |
|3  |1   |0   |0   |
+---+----+----+----+

aggCols = map(lambda c: (first(col(c)) - last(col(c))).alias(c),filter(lambda c: c != "id", df.columns))

df.groupBy(col("id")).agg(*aggCols).show()
+---+----+----+----+
| id|val1|val2|val3|
+---+----+----+----+
|  1|  -1|  -3|  -5|
|  3|   1|   0|   0|
|  2|  -3|   1|   4|
+---+----+----+----+

【讨论】：

【解决方案2】：

让我们注册一个 udf 并使用 numpy 的 .ptp

from pyspark.sql import Window
import pyspark.sql.functions as F

c = udf(lambda x: float(np.ptp(x)), FloatType())#register udf
df.groupBy('id').agg(c(F.collect_list('val1')).alias('v1'),c(F.collect_list('val2')).alias('v2'),c(F.collect_list('val3')).alias('v3')).show()#Apply udf

+---+---+---+---+
| id| v1| v2| v3|
+---+---+---+---+
|  1|1.0|3.0|5.0|
|  2|3.0|1.0|4.0|
|  3|1.0|0.0|0.0|
+---+---+---+---+

【讨论】：