在Spark中,可以使用groupBy
和agg
函数进行数据聚合。以下是一些常见的聚合操作:
- COUNT:计算每个分组中的行数。
from pyspark.sql.functions import count # 假设有一个名为df的DataFrame,其中有一个名为"column_name"的列 aggregated_df = df.groupBy("column_name").agg(count("*"))
- SUM:计算每个分组中的列值之和。
from pyspark.sql.functions import sum # 假设有一个名为df的DataFrame,其中有一个名为"column_name"的列 aggregated_df = df.groupBy("column_name").agg(sum("column_name"))
- AVG:计算每个分组中的列值平均值。
from pyspark.sql.functions import avg # 假设有一个名为df的DataFrame,其中有一个名为"column_name"的列 aggregated_df = df.groupBy("column_name").agg(avg("column_name"))
- MAX:计算每个分组中的列值最大值。
from pyspark.sql.functions import max # 假设有一个名为df的DataFrame,其中有一个名为"column_name"的列 aggregated_df = df.groupBy("column_name").agg(max("column_name"))
- MIN:计算每个分组中的列值最小值。
from pyspark.sql.functions import min # 假设有一个名为df的DataFrame,其中有一个名为"column_name"的列 aggregated_df = df.groupBy("column_name").agg(min("column_name"))
- COUNT DISTINCT:计算每个分组中不同列值的数量。
from pyspark.sql.functions import countDistinct # 假设有一个名为df的DataFrame,其中有一个名为"column_name"的列 aggregated_df = df.groupBy("column_name").agg(countDistinct("column_name"))
这些示例仅展示了基本的聚合操作。你还可以使用agg
函数与多个聚合函数一起使用,例如:
from pyspark.sql.functions import sum, avg, count # 假设有一个名为df的DataFrame,其中有一个名为"column_name"的列 aggregated_df = df.groupBy("column_name").agg( sum("column_name"), avg("column_name"), count("*") )
此外,你还可以使用groupBy
函数的多个参数进行更复杂的聚合操作,例如:
# 假设有一个名为df的DataFrame,其中有两个名为"column1"和"column2"的列 aggregated_df = df.groupBy("column1", "column2").agg( sum("column_name"), avg("column_name"), count("*") )