如何在Databricks中计算多PySpark DataFrame间重复出现的distinct label数量

嗨，针对你的需求，我推荐先提取每个DataFrame的唯一label，再通过交集操作来计算不同组合的共有label数量——这对于数十亿行的大数据量来说是更高效的做法，避免处理重复数据浪费资源。下面是具体的实现步骤和代码：

第一步：提取各DataFrame的唯一Label #

首先从三个DataFrame中取出去重后的label列，减少后续计算的数据量：

# 提取每个DataFrame的distinct label
raw_old_labels = raw_old.select("label").distinct()
raw_new_labels = raw_new.select("label").distinct()
master_labels = master_df.select("label").distinct()

第二步：计算不同组合的共有Label数量 #

你可以用两种方式实现，选你觉得更直观的即可：

方法一：使用Join操作

通过内连接（inner join）获取两个DataFrame的共有label，再统计数量：

# 1. raw_old 和 raw_new 共有的label数量
old_new_common = raw_old_labels.join(raw_new_labels, on="label", how="inner")
old_new_count = old_new_common.count()
print(f"raw_old 和 raw_new 共有的label数量: {old_new_count}")  # 输出：3

# 2. raw_new 和 master_df 共有的label数量
new_master_common = raw_new_labels.join(master_labels, on="label", how="inner")
new_master_count = new_master_common.count()
print(f"raw_new 和 master_df 共有的label数量: {new_master_count}")  # 输出：2

# 3. raw_old 和 master_df 共有的label数量
old_master_common = raw_old_labels.join(master_labels, on="label", how="inner")
old_master_count = old_master_common.count()
print(f"raw_old 和 master_df 共有的label数量: {old_master_count}")  # 输出：4

# 4. 三个DataFrame都共有的label数量
all_common = raw_old_labels.join(raw_new_labels, on="label") \
                           .join(master_labels, on="label")
all_common_count = all_common.count()
print(f"三个DataFrame都共有的label数量: {all_common_count}")  # 输出：1

方法二：使用Intersect方法

PySpark内置的intersect()方法可以直接返回两个DataFrame的交集（自动去重），代码更简洁：

# 1. raw_old 和 raw_new 共有的label数量
old_new_count = raw_old_labels.intersect(raw_new_labels).count()
print(f"raw_old 和 raw_new 共有的label数量: {old_new_count}")  # 输出：3

# 2. raw_new 和 master_df 共有的label数量
new_master_count = raw_new_labels.intersect(master_labels).count()
print(f"raw_new 和 master_df 共有的label数量: {new_master_count}")  # 输出：2

# 3. raw_old 和 master_df 共有的label数量
old_master_count = raw_old_labels.intersect(master_labels).count()
print(f"raw_old 和 master_df 共有的label数量: {old_master_count}")  # 输出：4

# 4. 三个DataFrame都共有的label数量
all_common_count = raw_old_labels.intersect(raw_new_labels).intersect(master_labels).count()
print(f"三个DataFrame都共有的label数量: {all_common_count}")  # 输出：1

注意事项 #

因为你的实际数据是数十亿行，先提取distinct label是关键优化点——这一步能大幅减少后续计算的数据量，让分布式计算更高效，避免不必要的资源消耗。两种方法在性能上差异不大，你可以根据自己的代码习惯选择。

备注：内容来源于stack exchange，提问作者carousallie