本文介绍如何使用 PySpark 的内连接（inner join）高效统计多个大规模 DataFrame 在指定列（如 label）上的共同值数量，适用于 Databricks 环境下的海量数据场景。
本文介绍如何使用 pyspark 的内连接（inner join）高效统计多个大规模 dataframe 在指定列（如 `label`）上的共同值数量，适用于 databricks 环境下的海量数据场景。
在处理超大规模数据（如数十亿行）时，直接对 DataFrame 做 collect + Python 集合运算会导致内存溢出或严重性能退化。PySpark 提供了基于分布式计算的高效方案：利用 join(..., how="inner") 获取交集，再调用 .count() 获取交集元素个数——所有操作均在集群中完成，无需将数据拉回 Driver。
✅ 核心原理
inner join 仅保留左右表在连接键上完全匹配的行。因此，对两表按 label 列内连接后，结果行数即为二者 label 的交集基数（distinct count of common labels）。注意：即使某 label 在任一表中重复出现，只要至少各出现一次，就计入交集；.count() 统计的是匹配行总数，但因我们只关心“是否共存”，需确保 label 列在连接前已去重，否则会高估交集大小（例如 raw_old 中 A789 出现 2 次、raw_new 中出现 3 次，内连接将生成 6 行，但交集仍为 1 个 label）。因此，推荐先对每张表的 label 列做 distinct 去重再连接：
# 步骤 1：分别提取并去重 label 列
old_labels = raw_old.select("label").distinct()
new_labels = raw_new.select("label").distinct()
master_labels = master_df.select("label").distinct()
# 步骤 2：两两计算交集数量
old_and_new = old_labels.join(new_labels, on="label", how="inner").count()
print(f"raw_old ∩ raw_new: {old_and_new}") # 输出: 3
new_and_master = new_labels.join(master_labels, on="label", how="inner").count()
print(f"raw_new ∩ master_df: {new_and_master}") # 输出: 2
old_and_master = old_labels.join(master_labels, on="label", how="inner").count()
print(f"raw_old ∩ master_df: {old_and_master}") # 输出: 4
# 步骤 3：三表交集（链式 inner join）
old_new_master = (old_labels
.join(new_labels, on="label", how="inner")
.join(master_labels, on="label", how="inner")
.count())
print(f"raw_old ∩ raw_new ∩ master_df: {old_new_master}") # 输出: 1⚠️ 关键注意事项
-
务必先
.distinct()：避免因 label 重复导致 join 后行数膨胀，使.count()返回错误结果（它统计的是匹配行数，而非 distinct label 数）。 -
连接键一致性：确保所有表中
label列数据类型、空格、大小写完全一致；必要时添加.filter(col("label").isNotNull()).withColumn("label", trim(lower(col("label"))))进行标准化。 -
性能优化：对于超大表，可考虑对
label列进行广播（若去重后 label 总数 broadcast() 提升小表 join 效率：from pyspark.sql.functions import broadcast old_and_new = old_labels.join(broadcast(new_labels), "label").count()
- 扩展性：该模式可自然扩展至 N 个表——依次两两 inner join 即可，Spark 会自动优化执行计划。
通过上述方法，你能在秒级内完成数十亿行数据的多表 label 交集统计，兼顾准确性、可扩展性与生产环境稳定性。
