RDD - Transformation Join

The join() transformation is used to combine elements from two Key-Value Pair RDDs based on a shared common key. Spark supports all standard relational SQL join variants: Inner Join, Left Outer Join, Right Outer Join, and Full Outer Join.

The Shuffle Cost of Joins

Joining distributed datasets requires a Wide Transformation. Spark shuffles data across the network to ensure that all records matching the same key from both RDDs land on the exact same executor node to perform the final cross-product join.

PySpark Code Examples

Setup Spark Session

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("RDD Transformation Join") \
    .master("local[*]") \
    .getOrCreate()

sc = spark.sparkContext

Define Sample Datasets

We will use two key-value datasets representing Users and their Purchase Orders:

# RDD A: (UserID, Username)
users_rdd = sc.parallelize([(1, "Alice"), (2, "Bob"), (3, "Charlie")])

# RDD B: (UserID, PurchaseItem)
orders_rdd = sc.parallelize([(1, "Laptop"), (1, "Mouse"), (2, "Book")])

A. Inner Join (join)

Returns pairs of elements with keys present in both RDDs. If a key is missing from either, the record is excluded.

inner_joined = users_rdd.join(orders_rdd)

print("Inner Join Results:")
print(inner_joined.collect())
# Output: [(1, ('Alice', 'Laptop')), (1, ('Alice', 'Mouse')), (2, ('Bob', 'Book'))]
# Note: Charlie (UserID 3) is excluded because he has no purchase orders!

B. Left Outer Join (leftOuterJoin)

Returns keys present in the left (first) RDD. If a key is missing from the right (second) RDD, it fills the right side value with None.

left_joined = users_rdd.leftOuterJoin(orders_rdd)

print("
Left Outer Join Results:")
print(left_joined.collect())
# Output: [(1, ('Alice', 'Laptop')), (1, ('Alice', 'Mouse')), (2, ('Bob', 'Book')), (3, ('Charlie', None))]
# Note: Charlie is included, with his order represented as None!

C. Right Outer Join (rightOuterJoin)

Returns keys present in the right (second) RDD. If a key is missing from the left (first) RDD, the left side value is represented as None.

# Add an order with an ID that has no profile (orphan record)
orphan_orders_rdd = sc.parallelize([(1, "Laptop"), (4, "Keyboard")])

right_joined = users_rdd.rightOuterJoin(orphan_orders_rdd)

print("
Right Outer Join Results:")
print(right_joined.collect())
# Output: [(1, ('Alice', 'Laptop')), (4, (None, 'Keyboard'))]
# Note: Purchase 4 is included, with the username represented as None!

D. Full Outer Join (fullOuterJoin)

Returns keys present in either the left or the right RDD. Absent fields from either side are filled with None.

full_joined = users_rdd.fullOuterJoin(orphan_orders_rdd)

print("
Full Outer Join Results:")
print(full_joined.collect())
# Output: [(1, ('Alice', 'Laptop')), (2, ('Bob', None)), (3, ('Charlie', None)), (4, (None, 'Keyboard'))]