Scala Spark实现跨门店独有商品查找及所属门店名关联

• 计算商品出现频次时，完全丢弃了门店字段，统计结果仅能拿到商品本身，无法反向关联所属门店
• 硬编码取x(1)~x(5)的固定位置列，只要某门店的商品数量不足5个，直接触发数组越界异常；如果商品超过5个，会直接漏算末尾的商品
• 筛选出独有商品后需要额外和原表做join匹配门店，增加不必要的shuffle开销

整个思路从第一步数据解析就保留商品和门店的绑定关系，不需要后续反向匹配，同时解决硬编码列的适配问题：

1. 逐行解析原始数据，第一列识别为门店名，后续所有列作为该门店的售卖商品列表，不再写死列长度
2. 将每个商品展开为(商品名, 所属门店)的键值对结构
3. 按商品聚合，统计每个商品被多少个门店售卖，筛选出仅被1个门店售卖的独有商品
4. 按照要求格式拼接输出语句

完整可运行代码 #

val filerdd = sc.textFile("file:///home/hduser/stores_products")

// 解析原始行，生成(商品, 所属门店)的扁平结构
val productStoreRdd = filerdd.map(row => {
  val splitRow = row.split(",")
  val store = splitRow(0)
  // 从索引1开始取所有剩余元素作为商品，适配不同长度的行
  // 加distinct避免同门店重复录入同一商品导致计数错误
  val products = splitRow.drop(1).distinct
  products.map(product => (product, store))
}).flatMap(identity)

// 聚合统计并筛选独有商品
val uniqueProductRdd = productStoreRdd
  .groupByKey()
  // 仅保留仅在1个门店售卖的商品
  .filter { case (_, storeList) => storeList.size == 1 }
  .map { case (product, storeList) =>
    val targetStore = storeList.head
    // 商品名首字母大写适配输出格式
    val formattedProduct = product.capitalize
    s"Only $targetStore has $formattedProduct."
  }

// 触发执行并打印结果
uniqueProductRdd.collect().foreach(println)

大流量场景优化 #

如果业务数据量较大，可将聚合逻辑替换为带map端局部预聚合的reduceByKey实现，减少shuffle阶段的数据传输量，核心代码片段为：

val uniqueProductRdd = productStoreRdd
  .map{ case (product, store) => (product, (1, store)) }
  // 累加商品出现次数，保留任意一个关联门店即可
  .reduceByKey{ case ((cntA, storeA), (cntB, _)) => (cntA + cntB, storeA) }
  .filter{ case (_, (appearCount, _)) => appearCount == 1 }
  .map{ case (product, (_, store)) => s"Only $store has ${product.capitalize}." }

运行输出示例 #

Only Sears has Pen.
Only Walmart has Meat.

注：collect()操作不保证固定输出顺序，如需固定排序可调用sortBy按门店名排序后再打印即可。

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