我正在使用PySpark DataFrames分析一些数据。假设我有一个要聚合的数据帧df：
(df.groupBy("group")
.agg({"money":"sum"})
.show(100)
)
这将给我：
group SUM(money#2L)
A 137461285853
B 172185566943
C 271179590646
聚合工作正常，但我不喜欢新的列名SUM(money#2L)。有没有办法将此列重命名为可从.agg方法读取的内容？也许更类似于人们在dplyr中会做的事情：
df %>% group_by(group) %>% summarise(sum_money = sum(money))
虽然我仍然更喜欢dplyr语法，但这个代码片段可以：
import pyspark.sql.functions as sf
(df.groupBy("group")
.agg(sf.sum('money').alias('money'))
.show(100))
它变得冗长。
withColumnRenamed应该可以解决问题。这是 pyspark.sql API 的链接。
df.groupBy("group")
.agg({"money":"sum"})
.withColumnRenamed("SUM(money)", "money")
.show(100)
我为此做了一个小的辅助函数，可能会帮助一些人。
import re
from functools import partial
def rename_cols(agg_df, ignore_first_n=1):
"""changes the default spark aggregate names `avg(colname)`
to something a bit more useful. Pass an aggregated dataframe
and the number of aggregation columns to ignore.
"""
delimiters = "(", ")"
split_pattern = '|'.join(map(re.escape, delimiters))
splitter = partial(re.split, split_pattern)
split_agg = lambda x: '_'.join(splitter(x))[0:-ignore_first_n]
renamed = map(split_agg, agg_df.columns[ignore_first_n:])
renamed = zip(agg_df.columns[ignore_first_n:], renamed)
for old, new in renamed:
agg_df = agg_df.withColumnRenamed(old, new)
return agg_df
举个例子：
gb = (df.selectExpr("id", "rank", "rate", "price", "clicks")
.groupby("id")
.agg({"rank": "mean",
"*": "count",
"rate": "mean",
"price": "mean",
"clicks": "mean",
})
)
>>> gb.columns
['id',
'avg(rate)',
'count(1)',
'avg(price)',
'avg(rank)',
'avg(clicks)']
>>> rename_cols(gb).columns
['id',
'avg_rate',
'count_1',
'avg_price',
'avg_rank',
'avg_clicks']
至少做一点事情来避免人们打字这么多。
很简单：
val maxVideoLenPerItemDf = requiredItemsFiltered.groupBy("itemId").agg(max("playBackDuration").as("customVideoLength"))
maxVideoLenPerItemDf.show()
在 agg 中使用.as来命名创建的新行。
如果您
愿意对列名进行硬编码，.alias和.withColumnRenamed都可以工作。如果您需要一个编程解决方案，例如，为所有剩余列的聚合提供更友好的名称，这提供了一个很好的起点：
grouping_column = 'group'
cols = [F.sum(F.col(x)).alias(x) for x in df.columns if x != grouping_column]
(
df
.groupBy(grouping_column)
.agg(
*cols
)
)
df = df.groupby('Device_ID').agg(aggregate_methods)
for column in df.columns:
start_index = column.find('(')
end_index = column.find(')')
if (start_index and end_index):
df = df.withColumnRenamed(column, column[start_index+1:end_index])
上面的代码可以去除"（）"之外的任何内容。例如，"sum（foo）"将重命名为"foo"。
import findspark
findspark.init()
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
spark = SparkSession.builder.appName('test').getOrCreate()
data = [(1, "siva", 100), (2, "siva2", 200),(3, "siva3", 300),(4, "siva4", 400),(5, "siva5", 500)]
schema = ['id', 'name', 'sallary']
df = spark.createDataFrame(data, schema=schema)
df.show()
+---+-----+-------+
| id| name|sallary|
+---+-----+-------+
| 1| siva| 100|
| 2|siva2| 200|
| 3|siva3| 300|
| 4|siva4| 400|
| 5|siva5| 500|
+---+-----+-------+
**df.agg({"sallary": "max"}).withColumnRenamed('max(sallary)', 'max').show()**
+---+
|max|
+---+
|500|
+---+
虽然之前给出的答案很好，但我认为它们缺乏一种简洁的方式来处理.agg()字典的使用
如果要使用字典（实际上也可能是动态生成的，因为您有数百列），则可以使用以下内容而无需处理数十个代码行：
# Your dictionary-version of using the .agg()-function
# Note: The provided logic could actually also be applied to a non-dictionary approach
df = df.groupBy("group")
.agg({
"money":"sum"
, "...": "..."
})
# Now do the renaming
newColumnNames = ["group", "money", "..."] # Provide the names for ALL columns of the new df
df = df.toDF(*newColumnNames) # Do the renaming
当然，newColumnNames列表也可以动态生成。 例如，如果仅将聚合中的列追加到df则可以预先存储newColumnNames = df.columns，然后仅附加其他名称。
无论如何，请注意，newColumnNames必须包含数据帧的所有列名，而不仅仅是要重命名的列名（因为.toDF()由于 Sparks 不可变 RDD 而创建一个新的数据帧）！
另一个快速的小衬里来添加混合：
df.groupBy('group')
.agg({'money':'sum',
'moreMoney':'sum',
'evenMoreMoney':'sum'
})
.select(*(col(i).alias(i.replace("(",'_').replace(')','')) for i in df.columns))
只需将别名函数更改为您想要命名的任何名称即可。sum_moreMoney，上面生成了sum_money，因为我确实喜欢在变量名称中看到运算符。