Spark高效Lookup优化咨询：4TB增量基表与3GB日数据ID匹配

1. 基于ID构建布隆过滤器（Bloom Filter） #

• 预计算基表所有proc_id的布隆过滤器，存储在HDFS或本地缓存中，每日更新时可追加新数据的ID到过滤器，或定期全量重建。
• 处理日数据时，先通过布隆过滤器过滤掉肯定不存在的ID，再对剩余疑似存在的ID做精确查询，大幅减少后续Join的数据量。
• Spark代码示例：

  import org.apache.spark.util.sketch.BloomFilter
  // 预构建布隆过滤器（可每日增量更新或定期全量生成）
  val bloomFilter = base_data.select("proc_id")
    .rdd.map(_.getLong(0))
    .aggregate(BloomFilter.create(1000000000, 0.01))(
      (bf, id) => { bf.put(id); bf },
      (bf1, bf2) => { bf1.merge(bf2); bf1 }
    )
  // 广播过滤器到Executor
  val broadcastBf = spark.sparkContext.broadcast(bloomFilter)
  // 过滤日数据
  val filteredDaily = daily_data.filter(row => {
    val cId = row.getLong(row.fieldIndex("c_id"))
    broadcastBf.value.mightContain(cId)
  })
  // 精确匹配
  val common_ids = base_data.join(filteredDaily, base_data.proc_id === filteredDaily.c_id, "inner")
  

2. 基表ID哈希分区+局部匹配 #

• 对基表按proc_id做哈希分区，固定分区数（比如1000，根据集群资源调整）并持久化。
• 处理日数据时，按c_id做相同哈希分区，每个分区仅对应基表的同分区数据做匹配，避免全量Shuffle。
• 关键操作：

  // 创建分区基表
  base_data.repartition(1000, col("proc_id"))
    .write.mode("append").bucketBy(1000, "proc_id")
    .saveAsTable("partitioned_base_table")
  // 日数据匹配
  val partitionedDaily = daily_data.repartition(1000, col("c_id"))
  val common_ids = spark.table("partitioned_base_table")
    .join(partitionedDaily, col("proc_id") === col("c_id"), "inner")
  

3. 用半连接（Semi Join）替代内连接 #

• 半连接只返回左表中在右表存在的记录，不会产生重复数据，Spark优化器对其有专门优化，执行逻辑比内连接更高效。
• 代码示例：

  // 直接获取存在匹配的基表记录
  val common_ids = base_data.join(daily_data, base_data.proc_id === daily_data.c_id, "semi")
  // 若仅需ID集合
  val common_id_set = base_data.select("proc_id")
    .join(daily_data.select("c_id"), col("proc_id") === col("c_id"), "semi")
  

4. 基表ID同步到分布式键值存储 #

• 将基表proc_id同步到Redis、HBase这类键值存储，构建ID索引。处理日数据时，Executor直接查询键值存储判断存在性，绕过Spark Shuffle。
• 代码示例（Redis）：

  // 预加载基表ID到Redis（每日增量追加）
  base_data.select("proc_id").rdd.foreachPartition(iter => {
    val jedis = new Jedis("redis-host", 6379)
    iter.foreach(row => jedis.sadd("base_ids", row.getLong(0).toString))
    jedis.close()
  })
  // 日数据过滤
  val common_ids = daily_data.filter(row => {
    val cId = row.getLong(row.fieldIndex("c_id")).toString
    val jedis = new Jedis("redis-host", 6379)
    val exists = jedis.sismember("base_ids", cId)
    jedis.close()
    exists
  })
  

5. 增量维护基表ID集合 #

• 每日将新增基表ID追加到增量ID表，同时每周/每月重建一次全量ID快照表。处理日数据时，分别匹配增量表和快照表，合并结果。
• 逻辑示例：

  // 存储当日新增基表ID
  val dailyBaseIncrement = base_data.filter(col("dt") === current_date())
  dailyBaseIncrement.select("proc_id").write.mode("append").saveAsTable("base_id_increment")
  // 匹配增量ID
  val matchIncrement = daily_data.join(dailyBaseIncrement, col("c_id") === col("proc_id"), "inner")
  // 匹配快照ID
  val matchSnapshot = daily_data.join(spark.table("base_id_snapshot"), col("c_id") === col("proc_id"), "inner")
  // 合并去重
  val common_ids = matchIncrement.union(matchSnapshot).distinct()
  

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