如何在Spark表中增量跟踪不同值，高效计算日期区间客户平均收入

需求梳理 #

我们有一张超大Spark表，结构和样本数据如下：

核心需求是计算任意日期区间的单客户平均收入：区间总交易金额 ÷ 区间内有交易的唯一客户数。比如：

• 2022-12-20 ~ 2022-12-21：总金额75，唯一客户2位，平均34.5
• 2022-12-20 ~ 2022-12-22：总金额90，唯一客户3位，平均28

当前方案是存储每日客户集合，查询时合并计算大小，但客户量极大时会导致存储爆炸、查询效率极低，下面给出两种更优的实现方式。

方案1：近似统计——用HyperLogLog（HLL）实现高效增量维护 #

Spark原生支持HyperLogLog这种基数估计算法，能以极小的存储空间（KB级别）近似统计唯一值数量，误差通常在1%以内，完全适配海量客户场景。

具体步骤 #

1. 每日增量汇总
   每天运行Pipeline时，对当日数据计算两个指标并存入汇总表：

   • 当日交易总金额（daily_total）
   • 当日活跃客户的HLL结构（daily_hll）

   汇总表结构示例：

   Date	daily_total	daily_hll
   2022-12-20	42	HLL(Mary, Bob)
   2022-12-21	27	HLL(Mary, Bob)
   2022-12-22	15	HLL(Alice)

2. 区间查询计算
   查询任意日期区间时：

   • 累加区间内所有daily_total得到总金额
   • 合并区间内所有daily_hll，调用hll_count_merge函数得到近似唯一客户数
   • 总金额除以唯一客户数就是平均收入

   Spark SQL示例：

   SELECT
     '2022-12-20' AS start_date,
     '2022-12-21' AS end_date,
     SUM(daily_total) AS total_amount,
     hll_count_merge(daily_hll) AS unique_customers,
     ROUND(SUM(daily_total) / hll_count_merge(daily_hll), 2) AS avg_income_per_customer
   FROM daily_summary
   WHERE date BETWEEN '2022-12-20' AND '2022-12-21'
   

优势 #

• 存储成本极低：每个HLL仅占几十KB，百万级客户也不会有存储压力
• 查询效率极高：HLL合并是O(1)操作，区间计算秒级完成
• 误差可控：默认配置下误差小于1%，完全满足多数业务分析需求

方案2：精确统计——用RoaringBitmap实现高效压缩存储 #

如果业务要求100%精确的唯一客户数，可使用RoaringBitmap（高效压缩Bitmap实现），Spark生态中有第三方库支持（比如org.roaringbitmap:RoaringBitmapSpark）。

具体步骤 #

1. 每日增量生成Bitmap
   每天对当日活跃客户生成RoaringBitmap，同时计算当日总金额，存入汇总表：

   汇总表结构示例：

   Date	daily_total	customer_bitmap
   2022-12-20	42	Bitmap(Mary,Bob)
   2022-12-21	27	Bitmap(Mary,Bob)
   2022-12-22	15	Bitmap(Alice)

2. 区间查询计算
   查询时合并区间内的所有Bitmap，调用cardinality()方法得到精确的唯一客户数，再结合总金额计算平均收入。

   Scala代码示例：

   import org.roaringbitmap.RoaringBitmap
   
   val dailySummary = spark.table("daily_summary")
   val intervalData = dailySummary.filter($"date".between("2022-12-20", "2022-12-21"))
   
   // 计算区间总金额
   val totalAmount = intervalData.agg(sum("daily_total")).first().getLong(0)
   // 合并区间内的Bitmap
   val mergedBitmap = intervalData
     .agg(collect_list("customer_bitmap"))
     .first()
     .getList(0)
     .asScala
     .foldLeft(RoaringBitmap.empty())((acc, bitmap) => {
       acc.or(bitmap.asInstanceOf[RoaringBitmap])
       acc
     })
   // 得到精确唯一客户数并计算平均收入
   val uniqueCustomers = mergedBitmap.getCardinality
   val avgIncome = totalAmount.toDouble / uniqueCustomers
   

优势 #

• 完全精确：无误差的唯一客户数统计
• 存储高效：RoaringBitmap采用压缩存储，百万级客户仅需几MB
• 合并快速：Bitmap的OR操作效率远高于集合合并

方案对比 #

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