PySpark中如何补全groupBy后缺失的分组组合？

PySpark没有直接对应R dplyr中complete或expand的内置函数，但可以通过组合内置API实现相同效果，无需手动构建全量分组组合，尤其适合多分组列的场景。

核心思路 #

1. 提取所有分组列的唯一值集合，生成笛卡尔积得到全量可能的分组组合
2. 将全量组合与原分组统计结果做左连接，把缺失的count填充为0

基础示例（date + hour分组） #

假设原DataFrame df 包含date（日期）和hour（小时）列，先做基础统计：

from pyspark.sql import functions as F

# 原分组统计（会缺失无数据的date+hour组合）
grouped_df = df.groupBy("date", "hour").agg(F.count("*").alias("count"))

生成全量分组组合并补全缺失值：

# 获取各分组列的唯一值集合
date_df = df.select("date").distinct()
hour_df = df.select("hour").distinct()

# 生成所有date+hour的笛卡尔积（全量组合）
full_combinations = date_df.crossJoin(hour_df)

# 左连接并填充缺失的count为0
result_df = full_combinations.join(grouped_df, on=["date", "hour"], how="left") \
    .withColumn("count", F.coalesce(F.col("count"), F.lit(0)))

多分组列场景（动态处理） #

如果分组列数量较多（如date/hour/category/region），可以通过动态逻辑批量处理：

from functools import reduce
from pyspark.sql import DataFrame

# 定义所有分组列
group_cols = ["date", "hour", "category", "region"]

# 批量生成每个分组列的唯一值DataFrame
distinct_dfs = [df.select(col).distinct() for col in group_cols]

# 批量做笛卡尔积生成全量组合
def cross_join_all(dfs):
    return reduce(lambda df1, df2: df1.crossJoin(df2), dfs)

full_combinations = cross_join_all(distinct_dfs)

# 原分组统计
grouped_df = df.groupBy(group_cols).agg(F.count("*").alias("count"))

# 左连接补0
result_df = full_combinations.join(grouped_df, on=group_cols, how="left") \
    .withColumn("count", F.coalesce(F.col("count"), F.lit(0)))

特殊场景：补全连续日期 #

如果需要补全原数据中不存在的连续日期（比如某几天完全无数据），可以用sequence函数生成连续日期序列：

# 生成2023-07-01至2023-07-31的所有连续日期
date_df = spark.sql("""
    SELECT sequence(to_date('2023-07-01'), to_date('2023-07-31'), interval 1 day) as date
""").select(F.explode("date").alias("date"))

# 再和hour的唯一值做笛卡尔积，得到全量日期+小时组合
hour_df = df.select("hour").distinct()
full_combinations = date_df.crossJoin(hour_df)

# 后续左连接补0步骤同上

内容的提问来源于stack exchange，提问作者Arturo Sbr