04.5 PySpark Joins

Architect goal: Understand joins as both a data-modeling operation and a distributed execution problem. You should be able to predict join cardinality, choose an execution strategy, diagnose duplicate explosions and skew, and read the physical plan.

1. Why joins are critical

A join combines rows from two datasets using a matching condition.

Typical enterprise examples:

• Orders + customers
• Flights + airport master
• Sales + product hierarchy
• SAP transactions + master/reference data
• Fact + dimension
• CDC stream + target table

Basic syntax:

from pyspark.sql import functions as F

result = orders.join(
    customers,
    orders.customer_id == customers.customer_id,
    "inner"
)

A join can be logically simple while being computationally expensive because Spark may redistribute data across the cluster.

2. First principle: understand cardinality

Before writing a join, define the relationship.

Relationship	Meaning
1:1	One row on each side
1:N	One left row can match many right rows
N:1	Many left rows match one right row
N:N	Many rows can match many rows

Never assume a supposed dimension key is unique. Validate it:

customers.groupBy("customer_id").count().filter("count > 1")

If the dimension contains duplicates, a fact-to-dimension join can multiply rows.

3. Join output cardinality

Consider:

Orders
customer_id | amount
A           | 100
A           | 200

Customer
customer_id | segment
A           | Retail
A           | VIP

An inner join produces four rows.

A subsequent SUM(amount) becomes 600 instead of the intended 300.

This is a data correctness defect, not merely a performance issue.

4. Join types

Common PySpark join types include:

• inner
• left
• right
• full / full_outer
• cross
• left_semi
• left_anti

orders.join(customers, "customer_id", "left")

5. Inner, left, right and full joins

Inner returns matching rows from both datasets.

Left preserves every left row and enriches matching rows from the right.

Right preserves every right row.

Full outer preserves rows from both sides and is particularly useful for reconciliation.

Do not choose a join type from habit. Translate the business requirement into row-preservation semantics first.

6. Semi join

A left-semi join returns left rows where a match exists on the right.

eligible_orders = orders.join(
    active_customers.select("customer_id").distinct(),
    "customer_id",
    "left_semi"
)

Think:

“Does a matching key exist?”

It does not add right-side columns.

7. Anti join

A left-anti join returns left rows for which no matching right row exists.

orphan_orders = orders.join(
    customers.select("customer_id").distinct(),
    "customer_id",
    "left_anti"
)

Common uses:

• Data-quality checks
• Missing master records
• Incremental processing
• Reconciliation
• Delete detection

8. Cross join

A cross join creates a Cartesian product.

For 1,000,000 rows × 10,000 rows, the theoretical result is 10 billion rows.

Use it intentionally and understand the resulting cardinality.

9. Composite join keys

Many enterprise joins require multiple columns.

result = fact.join(
    dimension,
    (fact.company_code == dimension.company_code) &
    (fact.fiscal_year == dimension.fiscal_year),
    "left"
)

Always validate whether the combination is unique. Checking only one component of a composite key is insufficient.

10. Join conditions

Joins can use:

• Equality
• Multiple equality predicates
• Expressions
• Range conditions
• NULL-safe equality

Example:

result = left.join(
    right,
    (left.start_date <= right.event_date) &
    (right.event_date <= left.end_date),
    "left"
)

Range joins can have very different performance characteristics from equi-joins.

11. NULL and joins

Normal equality does not consider NULL equal to NULL.

left.key == right.key

does not match two NULL keys.

For explicit NULL-safe equality:

left.key.eqNullSafe(right.key)

Use this only when the business semantics define NULL as a match.

Do not use NULL-safe equality merely to make a join “work.”

12. Make the output schema intentional

Use aliases and select only required columns:

o = orders.alias("o")
c = customers.alias("c")

result = (
    o.join(
        c,
        F.col("o.customer_id") == F.col("c.customer_id"),
        "left"
    )
    .select(
        "o.order_id",
        "o.customer_id",
        "o.amount",
        "c.segment"
    )
)

This prevents accidental duplicate columns and makes downstream contracts clearer.

13. Logical join vs physical join

You write:

orders.join(customers, "customer_id")

but Spark chooses an execution strategy using the logical plan, statistics, configuration, runtime capabilities, and data characteristics.

Common physical strategies include:

• Broadcast hash join
• Sort-merge join
• Shuffled hash join
• Nested-loop variants for some conditions

The logical operation and physical implementation are different layers.

14. Broadcast join

If one side is sufficiently small, Spark can broadcast it to executors.

Large dataset
    │
    ├──────────────┐
    │              │
    ▼              ▼
Executor 1      Executor 2 ...
    ▲              ▲
    └── broadcast small table ──┘

Example:

from pyspark.sql.functions import broadcast

result = fact.join(
    broadcast(customer_dim),
    "customer_id",
    "left"
)

Broadcast can avoid shuffling the large fact side.

Broadcast is not automatically faster

It has costs:

• Driver-side preparation
• Network distribution
• Executor memory
• Serialization
• Failure risk if the supposedly small side is not actually small

Use it deliberately and measure.

15. Automatic broadcast

Inspect the current setting:

spark.conf.get("spark.sql.autoBroadcastJoinThreshold")

Do not design architecture around a hard-coded threshold. Effective behavior depends on runtime, statistics, configuration, and workload.

Explicit broadcast hints are useful when data characteristics are known and stable.

16. Sort-merge join

A sort-merge join is a common strategy for large equi-joins.

Left input                 Right input
    ↓                          ↓
Partition by key          Partition by key
    ↓                          ↓
   Sort                       Sort
    \                        /
       ---- Merge ----

Important costs can include:

• Shuffle
• Sort
• Network transfer
• Disk spill
• CPU

For large tables, this can be appropriate.

17. Shuffled hash join

A shuffled hash join redistributes data by key and uses hash-based matching within partitions.

It can be useful for suitable data distributions and sizes, but should not be selected blindly.

Always inspect the physical plan.

18. Join strategy decision tree

Is one side genuinely small?
        │
      yes ──→ Broadcast candidate
        │
       no
        ↓
Large equi-join?
        │
      yes ──→ Shuffle-based strategy
        │
       no
        ↓
Non-equi / range condition?
        │
      yes ──→ Inspect specialized execution
        │
       no
        ↓
Investigate physical plan

This is a decision aid, not a substitute for measurement.

19. Explain the join

Run:

result.explain("formatted")

Look for operators such as:

• BroadcastHashJoin
• SortMergeJoin
• ShuffledHashJoin
• Exchange
• Sort

Exact operators depend on Spark version, runtime, statistics, and query shape.

20. Filter and project before a large join

Filter as early as practical:

active_orders = orders.filter(
    F.col("status") == "COMPLETED"
)

customer_dim = customers.select(
    "customer_id",
    "segment",
    "country"
)

result = active_orders.join(
    customer_dim,
    "customer_id",
    "left"
)

This reduces unnecessary data entering the join. Spark may also push predicates where safe, so inspect the physical plan rather than assuming source-code order determines execution.

21. Join skew

Suppose one customer owns 50% of all fact rows.

A shuffle join can send a disproportionate amount of data to one partition.

Symptoms:

• One task runs much longer.
• Shuffle read is highly uneven.
• Spill is concentrated.
• Stage waits on a straggler.

Join performance is often a data-distribution problem.

22. AQE and skew handling

Adaptive Query Execution can change execution using runtime statistics. Depending on runtime and configuration, AQE can help with:

• Coalescing post-shuffle partitions
• Converting some join strategies
• Handling skewed joins

AQE is not magic. A poor data model or extreme skew may still require query or data-layout changes.

23. Join hints

Hints can influence strategy selection:

from pyspark.sql.functions import broadcast

result = fact.join(
    broadcast(dim),
    "customer_id"
)

A hint is a directive, not proof of performance. Validate the resulting plan and runtime.

24. Detect duplicate explosion

Before the join:

fact.count()

dim.groupBy("key").count().filter("count > 1")

After the join:

joined.count()

If row count unexpectedly increases, investigate cardinality.

Do not “fix” the problem by blindly calling distinct().

25. Why blind distinct() is dangerous

joined.distinct()

may hide a broken join and can introduce another expensive shuffle.

Correct approach:

1. Find why rows multiplied.
2. Correct the key or relationship.
3. Deduplicate intentionally only if the business model requires it.

26. SAP-to-Databricks example

Suppose an SAP-derived fact contains sales document/item/company code and a reference layer contains customer, company, and product attributes.

Before joining:

• Establish source grain.
• Check master-data uniqueness.
• Determine 1:1 versus N:1.
• Select only required columns.
• Filter early.
• Determine whether the dimension is broadcastable.
• Validate post-join row counts.

This is the same reasoning used for BW/4HANA-style fact/dimension enrichment.

27. Production enrichment pattern

from pyspark.sql import functions as F
from pyspark.sql.functions import broadcast

fact = (
    spark.table("main.silver.sales")
    .filter(F.col("status") == "COMPLETED")
    .select("order_id", "customer_id", "amount")
)

dim = (
    spark.table("main.silver.customer")
    .select("customer_id", "segment", "country")
)

result = fact.join(
    broadcast(dim),
    "customer_id",
    "left"
)

Only use broadcast after validating that the dimension is genuinely small enough for the workload.

28. Production troubleshooting workflow

Step 1 — Logical correctness

• Is the join key correct?
• Is the relationship correct?
• Are duplicates expected?

Step 2 — Cardinality

• Input rows
• Distinct keys
• Post-join rows

Step 3 — Physical plan

result.explain("formatted")

Step 4 — Spark UI

Inspect:

• Exchange
• Shuffle read/write
• Task-duration distribution
• Spill
• Stragglers

Step 5 — Skew

Look for heavily concentrated keys.

Step 6 — Tune

Consider filtering, projection, broadcast, AQE, data layout, and join-key design.

29. Hands-on lab — cardinality

Create:

• 10 million orders
• 100,000 customers

Then intentionally duplicate some customer keys.

Measure:

orders.count()
customers.count()
joined.count()

Identify why the join result changes.

30. Hands-on lab — broadcast vs shuffle

Run:

fact.join(dim, "customer_id")

and:

fact.join(broadcast(dim), "customer_id")

Compare:

• Physical plan
• Exchange operators
• Runtime
• Shuffle
• Executor memory

Do not assume broadcast wins; measure.

31. Hands-on lab — semi vs inner

Compare:

orders.join(customers, "customer_id", "inner")

with:

orders.join(
    customers.select("customer_id").distinct(),
    "customer_id",
    "left_semi"
)

Use the semi join when you only need existence filtering.

32. Hands-on lab — anti join

Find orphan records:

orders.join(
    customers.select("customer_id").distinct(),
    "customer_id",
    "left_anti"
)

Turn this into a production data-quality check.

33. Architect decision matrix

Situation	Starting strategy
Small reference + huge fact	Broadcast candidate
Two large equi-joins	Shuffle-based strategy
Only existence required	Semi join
Only non-existence required	Anti join
Reconciliation	Full outer join
Potential duplicate dimension	Validate uniqueness first
Severe skew	Inspect distribution + AQE/skew strategy
Range join	Analyze separately from equi-join
Unknown behavior	Inspect physical plan + Spark UI

34. Interview questions

1. What are the different PySpark join types?
2. What is the difference between inner and left join?
3. What is a semi join?
4. What is an anti join?
5. Why can a join unexpectedly increase row count?
6. Broadcast join vs sort-merge join?
7. When should you avoid broadcast?
8. What is join skew?
9. How does AQE help skewed joins?
10. How do you troubleshoot a 2-hour join?
11. Why is distinct() after a join often a bad fix?
12. How would you join a 20 TB fact table with a 5 GB dimension?
13. How do you validate an N:1 relationship?
14. How do NULL keys behave?
15. How would you design SAP-to-Databricks enrichment joins?

35. Cheat sheet

# Inner
left.join(right, "key", "inner")

# Left
left.join(right, "key", "left")

# Semi
left.join(right.select("key").distinct(), "key", "left_semi")

# Anti
left.join(right.select("key").distinct(), "key", "left_anti")

# Broadcast
from pyspark.sql.functions import broadcast
left.join(broadcast(right), "key", "left")

# Composite key
left.join(
    right,
    (left.company_code == right.company_code) &
    (left.fiscal_year == right.fiscal_year),
    "left"
)

# NULL-safe
left.join(
    right,
    left.key.eqNullSafe(right.key),
    "left"
)

# Physical plan
result.explain("formatted")

36. Key takeaways

1. A join is both a logical relationship and a distributed execution problem.
2. Always understand cardinality before optimizing.
3. Duplicate dimension keys can silently corrupt aggregates.
4. Broadcast can eliminate a large shuffle, but consumes executor memory.
5. Large joins commonly involve Exchange/shuffle and possibly sort.
6. Semi and anti joins are powerful existence operations.
7. NULL semantics must be deliberate.
8. Skew creates stragglers and uneven resource consumption.
9. Blind distinct() can hide a broken data model.
10. The physical plan and Spark UI are the source of truth for performance.

Next

04.6 PySpark Window Functions — partitioning, ordering, ranking, running totals, lag/lead, frames, session-style patterns, performance, and enterprise analytics.

Official references

• PySpark DataFrame.join
• PySpark Column.eqNullSafe
• Spark SQL join hints
• Spark SQL performance tuning
• Databricks join hints