From 60 Minutes to 4: Optimizing Spark MERGE INTO on a 2 Billion Row Iceberg Table

We run a daily pipeline that upserts ~50 million records into an Iceberg table with over 2 billion rows, powered by Lakekeeper — our open-source REST Iceberg catalog. This table is our source of truth for user attributes — latest seen timestamps, device IDs, and other properties that downstream teams depend on. It needs to be fresh every day.

The problem: MERGE INTO on a table this size was taking about an hour. Here’s how we got it down to under 4 minutes.

The Baseline: 60 Minutes

The naive approach is straightforward — merge the daily incremental data into the target table:

MERGE INTO visits.visits_distinct AS target
USING daily_incremental AS source
ON target.ad_id = source.ad_id
WHEN MATCHED THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT ...

On a 2 billion row table with 50 million new records per day, Spark has to shuffle both sides on ad_id, sort them, and then execute the merge. That’s a lot of data moving across the network — and it was taking about an hour.

Optimization 1: Storage Partition Join → 40 Minutes

Our target table is already bucketed by ad_id into 256 buckets. The data is physically pre-organized on disk by the join key. Spark can take advantage of this with a storage partition join — instead of shuffling the massive target side, it reads each bucket directly. The key enabler is spark.sql.sources.v2.bucketing.shuffle.enabled — this tells Spark to shuffle the smaller side (50M rows) according to the partitioning reported by the larger side (256 buckets), so only the incremental data moves across the network.

# Enable storage-aware bucketing (Iceberg v2 tables)
spark.sql.sources.v2.bucketing.enabled=true
spark.sql.sources.v2.bucketing.pushPartValues.enabled=true
spark.sql.sources.v2.bucketing.shuffle.enabled=true

This alone dropped the runtime from 60 to ~40 minutes. The shuffle was no longer the bottleneck — but sorting still was.

Optimization 2: Shuffle Hash Join → 4 Minutes

Even with 256 buckets, Spark was still sorting both sides within each bucket before the merge. That’s sorting ~8 million rows per bucket on the target side, 256 times over. Sorting was the new bottleneck.

The fix: force a shuffle hash join on the smaller (incremental) side. A hash join builds a hash table from the smaller side and probes it with the larger side — no sorting required.

MERGE INTO visits.visits_distinct AS target
USING /*+ SHUFFLE_HASH(source) */ daily_incremental AS source
ON target.ad_id = source.ad_id
WHEN MATCHED THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT ...

# Prefer hash join over sort merge
spark.sql.join.preferSortMergeJoin=false

# Match shuffle partitions to bucket count
spark.sql.shuffle.partitions=256

This eliminated the sort on the 2 billion row side entirely. Runtime dropped to ~4 minutes.

The Gotcha: Spark Doesn’t Always Listen

Here’s the part that took the longest to figure out. The shuffle hash hint worked — sometimes. Other days, the pipeline would revert to 40 minutes because Spark silently chose sort merge join instead.

Spark is the final arbiter of join strategy. Even with a hint, it does a safety check: will the smaller side’s partition fit in memory? If Spark estimates it won’t, it falls back to sort merge join. This estimation uses spark.sql.autoBroadcastJoinThreshold — not just for broadcast joins, but also as a heuristic for whether a shuffle hash join is safe.

We tried disabling Adaptive Query Execution (AQE) alone, thinking it was re-optimizing the plan at runtime. That didn’t help. The fix was both:

# Disable AQE to prevent runtime re-optimization
spark.sql.adaptive.enabled=false

# Set threshold high enough so Spark trusts the hash join is safe
spark.sql.autoBroadcastJoinThreshold=512MB

The autoBroadcastJoinThreshold tells Spark: “a partition of the smaller table up to this size can fit in memory.” Since our daily incremental data across 256 buckets works out to ~200MB per partition, setting this to 512MB gives enough headroom for Spark to consistently pick shuffle hash join.

With both settings in place, the pipeline runs under 4 minutes every single time.

Summary

The takeaway: on bucketed Iceberg tables, the combination of storage partition joins and shuffle hash hints can be transformative. But Spark’s join strategy selection has subtle heuristics — you need to understand what’s happening under the hood to make it stick.