如何用PySpark判断产品反馈数据集中是否存在长尾现象？

我正在处理一个200MB的用户产品反馈数据库，每条记录对应一条用户反馈，包含不同年份的数据。数据库核心字段如下：

• asin：产品唯一标识符

需要验证长尾现象：判断任意年份中，是否20%的产品（按asin统计）贡献了超过65%的总反馈量。

我尝试用PySpark RDD做了初步聚合，代码如下：

reviews_rdd = reviews_df_split_date.rdd
print(type(reviews_rdd))

reviews_rdd\
.map(lambda r: ((r.asin, r.Year),1))\
.reduceByKey(lambda v1, v2: v1 + v2)\
.collect()

但卡在后续步骤：不知道如何按年份分组产品、筛选出反馈量Top的产品，最终验证长尾结论。

推荐使用PySpark DataFrame API实现（比RDD更简洁高效，适合结构化数据处理），步骤如下：

步骤1：按年份+产品聚合反馈量 #

先统计每年每个产品的总反馈数：

# 按年份和asin分组，统计每个产品每年的反馈量
product_year_reviews = reviews_df_split_date.groupBy("Year", "asin")\
    .count()\
    .withColumnRenamed("count", "review_count")

步骤2：按年份计算全局统计量 #

统计每年的总产品数、总反馈数，以及20%产品对应的数量阈值：

from pyspark.sql import functions as F

year_stats = product_year_reviews.groupBy("Year")\
    .agg(
        F.count("asin").alias("total_products"),  # 当年总产品数
        F.sum("review_count").alias("total_reviews"),  # 当年总反馈数
        F.round(F.count("asin") * 0.2).cast("int").alias("top_20pct_threshold")  # 20%产品的数量
    )

步骤3：按年份对产品按反馈量降序排序并计算累积占比 #

对每年的产品按反馈量从高到低排序，计算每个产品的反馈量占当年总反馈的比例，以及累积占比：

from pyspark.sql.window import Window

# 给每个年份内的产品按反馈量降序排名，计算占比和累积占比
ranked_products = product_year_reviews.join(year_stats, on="Year", how="inner")\
    .orderBy("Year", F.desc("review_count"))\
    .withColumn(
        "review_pct",
        F.col("review_count") / F.col("total_reviews")  # 单个产品反馈占比
    )\
    .withColumn(
        "cumulative_pct",
        F.sum("review_pct").over(Window.partitionBy("Year").orderBy(F.desc("review_count")))  # 累积占比
    )

步骤4：筛选Top20%产品的累积占比，验证长尾现象 #

取每个年份中前20%产品的最大累积反馈占比，判断是否超过65%：

# 按年份取前top_20pct_threshold个产品的最大累积占比，验证长尾
longtail_verification = ranked_products.withColumn(
    "rank",
    F.row_number().over(Window.partitionBy("Year").orderBy(F.desc("review_count")))
)\
.filter(F.col("rank") <= F.col("top_20pct_threshold"))\
.groupBy("Year")\
.agg(
    F.max("cumulative_pct").alias("top_20pct_cumulative_pct"),
    F.max("total_products").alias("total_products"),
    F.max("total_reviews").alias("total_reviews")
)\
.withColumn(
    "has_longtail",
    F.when(F.col("top_20pct_cumulative_pct") > 0.65, True).otherwise(False)
)

# 查看最终验证结果
longtail_verification.show()

关键说明 #

• 优先用DataFrame API：结构化API自带Catalyst优化器，处理200MB数据比RDD更高效，代码可读性也更强。
• 窗口函数Window.partitionBy("Year")是核心：实现按年份分组排序、累积占比计算，精准定位Top20%产品的反馈贡献。
• 最终结果的has_longtail字段会直接标记对应年份是否符合目标长尾现象。

内容的提问来源于stack exchange，提问作者12major e