python—将可变列数传递给pyspark.agg()

hmtdttj4 于 2021-05-19 发布在 Spark

关注(0)|答案(2)|浏览(696)

在pyspark中，我需要在运行时构建一个列列表，然后在这些列上进行聚合。
这是一个有效的例子：

T.agg(col1,col2)

但是我需要传递一个列列表类型list of column（而不是string）
不起作用：

cols=[col1,col2]   
T.agg(cols)     #Passes param as tupple1(List)

不起作用：

cols=[col1,col2]    
T.agg(tupple(cols))      #Passes param as tupple1(tupple2)

这两种情况在pyspark代码上都失败： assert all(isinstance(c, Column) for c in exprs), "all exprs should be Column" 问题是python将列表作为tuple1传递，我需要传递一个列列表。解决办法是什么？

2条答案

问题是您需要将列表解压为单个参数。您可以使用 * 接线员。

T.agg(*cols)

这个 * 解压列表中的元素。这里是另一个玩具的例子，它独立于pyspark。

def test(a, b, c, d):
    print(a, b, c, d)
data = [3, 4, 5, 6]
test(*data)
# Output:
# 3 4 5 6

根据注解，如果要进行“第一次”聚合，可以执行以下操作：

col_list = ['col1', 'col2']
exprs = [first(i).alias("first_"+i) for i in col_list]
df.groupBy("some_col").agg(*exprs).show()