PySpark分组统计count异常，求原因及正确实现方案

你的PySpark代码中第一个join的关联条件完全错误，这是导致thing_rate不一致的核心原因：

原代码中的join条件写的是：

(df_count_per_category.thing== df_count_per_category.thing) & (df_count_per_category.category== df_count_per_category.category)

这是将df_count_per_category的字段与自身做相等判断，这个条件永远为真，会导致原df的每一行都和df_count_per_category的所有行进行关联（笛卡尔积）。最终df_merge中会混入不属于当前行的thing_count值，虽然同一事物的行可能碰巧拿到正确的thing_count，但错误的关联会让部分行的thing_count对应其他事物的数值，计算出的thing_rate自然混乱。

有两种修正思路，推荐使用窗口函数（和你R代码的逻辑完全一致，更简洁高效）：

方案1：使用窗口函数（推荐，对齐R语言逻辑） #

直接用窗口函数按category和category,thing分组计算统计值，无需多次join：

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

# 按category和thing分组，计算每个事物的记录数
window_thing = Window.partitionBy("category", "thing")
# 按category分组，计算分类下总记录数
window_category = Window.partitionBy("category")

df_result = df.withColumn("thing_count", F.count("*").over(window_thing)) \
              .withColumn("thing_total", F.count("*").over(window_category)) \
              .withColumn("thing_rate", F.round(F.col("thing_count") / F.col("thing_total"), 4)) \
              .select("category", "thing", "thing_count", "thing_total", "thing_rate") \
              .distinct()  # 去重，保留唯一的category-thing组合

df_result.show()

方案2：修正join条件 #

如果坚持用join的方式，需要把第一个join的条件改成原df和df_count_per_category的字段关联：

# 原统计部分不变
df_count_per_category = df.select('category', 'thing') \
                          .groupby('category', 'thing') \
                          .agg(F.count('*').alias('thing_count'))

df_total = df.select('category') \
             .groupby('category') \
             .agg(F.count('*').alias('thing_total'))

# 修正第一个join的关联条件
df_merge = df.join(df_count_per_category, 
                   on=[df.category == df_count_per_category.category, 
                       df.thing == df_count_per_category.thing], 
                   how='inner') \
             .drop(df_count_per_category.thing) \
             .drop(df_count_per_category.category) \
             .join(df_total, on=[df.category == df_total.category], how='inner') \
             .drop(df_total.category)

df_rate = df_merge.withColumn('thing_rate', F.round(F.col('thing_count') / F.col('thing_total'), 4)) \
                  .select("category", "thing", "thing_count", "thing_total", "thing_rate") \
                  .distinct()

df_rate.show()

两种方案都会得到和你R代码一致的结果：

+---------+-----+-----------+------------+----------+
|category|thing|thing_count|thing_total|thing_rate|
+---------+-----+-----------+------------+----------+
|        A|    X|          3|           6|     0.5000|
|        A|    Y|          1|           6|     0.1667|
|        A|    Z|          2|           6|     0.3333|
|        B|    X|          1|           3|     0.3333|
|        B|    Y|          1|           3|     0.3333|
|        B|    Z|          1|           3|     0.3333|
+---------+-----+-----------+------------+----------+

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