How do I count distinct 'label' intersections across multiple PySpark DataFrames in Databricks?

Below is one way to compute the intersections using PySpark’s DataFrame API. The idea is to first select distinct labels from each DataFrame and then use the intersect() function to get the common rows (labels). Finally, count the number of rows in each resulting DataFrame.

Below is a complete example:

PYTHONCopied
# Select distinct labels from each DataFrame
raw_old_labels = raw_old.select("label").distinct()
raw_new_labels = raw_new.select("label").distinct()
master_labels  = master_df.select("label").distinct()

# Intersection between raw_old and raw_new
common_old_new = raw_old_labels.intersect(raw_new_labels)
print("Labels in raw_old and raw_new:", common_old_new.collect())  # Expected: A789, B456, D123
print("Count:", common_old_new.count())

# Intersection between raw_new and master_df
common_new_master = raw_new_labels.intersect(master_labels)
print("Labels in raw_new and master_df:", common_new_master.collect())  # Expected: C456, D123
print("Count:", common_new_master.count())

# Intersection between raw_old and master_df
common_old_master = raw_old_labels.intersect(master_labels)
print("Labels in raw_old and master_df:", common_old_master.collect())  # Expected: A654, B987, C987, D123
print("Count:", common_old_master.count())

# Intersection among raw_old, raw_new, and master_df
common_all = raw_old_labels.intersect(raw_new_labels).intersect(master_labels)
print("Labels in raw_old, raw_new, and master_df:", common_all.collect())  # Expected: D123
print("Count:", common_all.count())

Explanation

1. Extracting Distinct Labels:
   We use .select("label") along with .distinct() so that repeated labels (if any) don’t impact the intersection results.

2. Using intersect():
   The .intersect() method returns only those rows that exist in both DataFrames. You can chain .intersect() to compute the intersection among more than two DataFrames.

3. Counting the Results:
   The .count() method gives you the number of labels that appear in the resulting intersections.

This solution should scale well provided that you already have partitioned and optimized DataFrames, as Databricks’ Spark engine efficiently distributes the work over your cluster.