PySpark中生成交叉销售关联表的最优实现方案

需求回顾 #

给定包含customer_id和product_name的购买数据集，基于指定排序后的产品列表['hat','pants','shoes','tie']，生成交叉销售表：行代表某产品，列代表另一产品，单元格值为购买行产品的客户中同时购买列产品的总人数。

高效实现步骤 #

核心思路是先聚合每个客户的购买产品集合，减少后续计算的数据量，再通过自关联+透视完成交叉表构建，避免重复统计客户：

1. 初始化数据与指定产品列表
   先创建示例DataFrame，并定义排序后的目标产品列表：

   from pyspark.sql import SparkSession
   from pyspark.sql import functions as F
   
   # 初始化Spark会话
   spark = SparkSession.builder.appName("CrossSellTable").getOrCreate()
   
   # 原始数据
   data = [
       ("001", "shoes"),
       ("001", "tie"),
       ("001", "hat"),
       ("002", "shoes"),
       ("002", "tie")
   ]
   df = spark.createDataFrame(data, ["customer_id", "product_name"])
   
   # 指定排序后的产品列表
   product_list = sorted(['hat','pants','shoes','tie'])
   

2. 聚合客户的购买产品集合
   按客户ID分组，收集每个客户购买的所有产品（去重，同一客户多次买同一产品不影响关联统计）：

   customer_products = df.groupBy("customer_id")\
       .agg(F.collect_set("product_name").alias("products"))
   

3. 生成产品关联对
   保留客户ID的同时，对每个客户的产品集合做自关联，生成所有产品对（row_product, col_product），并过滤出仅属于目标产品列表的产品：

   product_pairs = customer_products.select(
       "customer_id",
       F.explode("products").alias("row_product"),
       F.explode("products").alias("col_product")
   ).filter(
       F.col("row_product").isin(product_list) & F.col("col_product").isin(product_list)
   )
   

4. 统计产品对的客户数
   按产品对分组，统计不同客户的数量（确保同一客户仅被计数一次）：

   pair_counts = product_pairs.groupBy("row_product", "col_product")\
       .agg(F.countDistinct("customer_id").alias("count"))
   

5. 透视生成交叉表
   使用pivot将列产品转为表头，按指定产品列表排序行和列，同时填充无关联的单元格为0：

   cross_sell_table = pair_counts.groupBy("row_product")\
       .pivot("col_product", product_list)\
       .agg(F.first("count"))\
       .na.fill(0)\
       .orderBy("row_product")
   

最终结果展示 #

执行cross_sell_table.show()后输出：

+------------+---+-----+-----+---+
|row_product|hat|pants|shoes|tie|
+------------+---+-----+-----+---+
|         hat|  1|    0|    1|  1|
|       pants|  0|    0|    0|  0|
|       shoes|  1|    0|    2|  2|
|         tie|  1|    0|    2|  2|
+------------+---+-----+-----+---+

完全符合需求：比如购买hat的客户（仅001）同时购买tie的人数是1，与描述一致。

效率说明 #

• 先聚合客户的产品集合，将原始N条购买记录压缩为M条客户记录（M远小于N），大幅减少后续计算的数据量；
• pivot时指定目标产品列表，避免Spark全量扫描所有可能的产品值，提升透视效率；
• 基于客户维度去重统计，避免重复计算同一客户的多次购买行为，保证结果准确且高效。

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