Debugging MERGE_CARDINALITY_VIOLATION in Apache Iceberg SCD Type 2 Pipelines on AWS Glue PySpark

By Henil Patel | AWS Certified Data Engineer | Data Engineer at Schellman

If you’re running SCD Type 2 MERGE operations on Apache Iceberg tables in AWS Glue PySpark and hitting MERGE_CARDINALITY_VIOLATION, the root cause is almost certainly xxhash64() in your hash-key generation creating duplicate logical rows in Spark’s query plan — even when your source data is clean. Here’s exactly how I diagnosed and fixed it in production.

The Setup: SCD Type 2 on Apache Iceberg in AWS Glue

At my company, we run a 4-layer medallion lakehouse on AWS — Bronze → Silver → Gold → Consumption — processing 50M+ records per month from sources including Workday, HubSpot, QuickBase, and Kantata. Our Silver layer implements SCD Type 2 (Slowly Changing Dimensions Type 2) to maintain full historical lineage of every record change.

The stack:

• AWS Glue 5.0 (PySpark 3.3 runtime)

• Apache Iceberg 1.4. (via — datalake-formats iceberg Glue config)

• AWS Lake Formation for table governance

• S3 as the storage layer

• Terraform for infrastructure provisioning

The SCD Type 2 pattern we use is standard: for each incoming batch, detect changed records, expire the current version (is_current = False, set end_date), and insert a new active version (is_current = True). We use a MERGE INTO statement to do this atomically on the Iceberg table.

The Silver job had been running fine in development. Then we promoted it to production — and it started exploding.

The Bug: What MERGE_CARDINALITY_VIOLATION Actually Means

The error appeared in Glue CloudWatch logs approximately 8 minutes into the job run:

AnalysisException: MERGE_CARDINALITY_VIOLATION
org.apache.spark.sql.AnalysisException:
MERGE_CARDINALITY_VIOLATION: The ON clause of the MERGE statement matched a single row from the target table with multiple rows of the source table.
at org.apache.spark.sql.errors.QueryCompilationErrors$.mergeCardinalityViolationError(QueryCompilationErrors.scala)
…

The Spark documentation says this error fires when a single target row matches more than one source row in a MERGE statement. The SQL standard for MERGE is strict: one target row can only be updated or deleted by exactly one source row. If there are duplicates in the source, Spark raises this exception rather than applying an arbitrary or nondeterministic update.

At face value, this looks like a data problem. Our immediate assumption: duplicate records in the incoming Bronze data. We were wrong.

The Investigation: Why the Source Data Wasn’t Actually Duplicated

Step 1 — Check the raw source data

We pulled the incoming batch from S3 and counted distinct records by natural key:

source_df = spark.read.format("parquet").load("s3://bucket/bronze/workday/workers/")
source_df.groupBy("worker_id").count().filter("count > 1").show()

Result: Zero duplicates. Every worker_id appeared exactly once in the source.

Step 2 — Check the Silver Iceberg table for duplicate active records

silver_df = spark.read.format("iceberg").load("glue_catalog.silver.dim_workers")
silver_df.filter("is_current = true") \
.groupBy("worker_id").count() \
.filter("count > 1").show()

Result: Zero duplicate active records. The target table was clean too.

So we had: clean source, clean target, and still a MERGE_CARDINALITY_VIOLATION. Something was being introduced inside the Spark query plan itself.

Step 3 — Inspect the hash key generation

Our SCD Type 2 logic generates a row_hash to detect changes between source and target:

from pyspark.sql.functions import xxhash64, col, concat_ws
source_df = source_df.withColumn(
"row_hash",
xxhash64(
concat_ws("||",
col("worker_id"),
col("first_name"),
col("last_name"),
col("department"),
col("cost_center"),
col("employment_status")
)
)
)

This looked innocent. But when we ran source_df.explain(extended=True), the physical plan revealed something unexpected.

Step 4 — The explain() that revealed everything

source_df.explain(extended=True)

In the logical plan output, we saw xxhash64 expanding into a non-deterministic expression subtree that Spark’s optimizer was treating as a source of potential row multiplication. Because xxhash64 uses an internal seed that Spark tracks through its expression lineage, the optimizer was conservatively marking certain plan nodes as potentially producing multiple output rows per input row.

The critical insight: Spark’s MERGE validation happens at the logical plan level, not after execution. It inspects the query plan to verify cardinality guarantees. If xxhash64 appears in the lineage of the JOIN key used in the MERGE ON clause, Spark cannot statically prove the one-to-one relationship — and raises the violation preemptively.

The specific trigger was that xxhash64 output was being used (indirectly, through a self-join in the change-detection logic) as part of the MERGE join condition. Spark saw a plan it couldn’t statically verify as cardinality-safe.

The Fix

Option 1 — Replace xxhash64 with md5 or sha2 (what we did)

from pyspark.sql.functions import md5, sha2, col, concat_ws
source_df = source_df.withColumn(
"row_hash",
md5(
concat_ws("||",
col("worker_id"),
col("first_name"),
col("last_name"),
col("department"),
col("cost_center"),
col("employment_status")
)
)
)

md5() and sha2() are deterministic in Spark’s expression framework. They don’t carry the internal seed lineage that xxhash64 does, so Spark’s optimizer can correctly reason about cardinality. The MERGE ran cleanly on the first attempt after this change.

sha2 with 256-bit output is recommended for production (lower collision probability than md5):

source_df = source_df.withColumn(
"row_hash",
sha2(concat_ws("||", col("worker_id"), col("first_name"),
col("last_name"), col("department"),
col("cost_center"), col("employment_status")), 256)
)

Option 2 — Deduplicate before MERGE with explicit window function

If you need to keep xxhash64 (e.g., for performance on very wide rows), add an explicit deduplication step before the MERGE to give Spark a static cardinality guarantee:

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, desc
window = Window.partitionBy("worker_id").orderBy(desc("effective_date"))
source_deduped = source_df \
.withColumn("rn", row_number().over(window)) \
.filter("rn = 1") \
.drop("rn")

This materialized deduplication — even when the data is already deduplicated — forces Spark to treat the source as cardinality-safe because the window function result is physically bounded.

Option 3 — Force a checkpoint to break lineage

source_df = source_df.checkpoint() # requires SparkContext.setCheckpointDir()

Checkpointing breaks the logical plan lineage entirely, so the xxhash64 expression tree is no longer visible to the MERGE validator. This is the most aggressive option and has I/O overhead — use only if Options 1 and 2 don’t apply.

The Full Corrected SCD Type 2 MERGE Pattern

For reference, here is the complete corrected pattern we use in production:

from pyspark.sql.functions import sha2, concat_ws, col, lit, current_timestamp
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
.config("spark.sql.catalog.glue_catalog", "org.apache.iceberg.spark.SparkCatalog") \
.config("spark.sql.catalog.glue_catalog.warehouse", "s3://your-bucket/warehouse/") \
.config("spark.sql.catalog.glue_catalog.catalog-impl", "org.apache.iceberg.aws.glue.GlueCatalog") \
.getOrCreate()

# 1. Load source
source_df = spark.read.format("parquet") \
.load("s3://your-bucket/bronze/workday/workers/")

# 2. Generate row hash - use sha2, NOT xxhash64
source_df = source_df.withColumn(
"row_hash",
sha2(concat_ws("||",
col("worker_id"), col("first_name"), col("last_name"),
col("department"), col("cost_center"), col("employment_status")
), 256)
).withColumn("effective_date", current_timestamp()) \
.withColumn("end_date", lit(None).cast("timestamp")) \
.withColumn("is_current", lit(True))

# 3. Register as temp view
source_df.createOrReplaceTempView("source_updates")

# 4. MERGE - expire changed records and insert new versions
spark.sql("""
MERGE INTO glue_catalog.silver.dim_workers AS target
USING (
SELECT
s.*,
CASE WHEN t.row_hash IS NULL THEN 'INSERT'
WHEN t.row_hash != s.row_hash THEN 'UPDATE'
ELSE 'NOCHANGE' END AS merge_action
FROM source_updates s
LEFT JOIN glue_catalog.silver.dim_workers t
ON s.worker_id = t.worker_id AND t.is_current = true
) AS staged
ON target.worker_id = staged.worker_id
AND target.is_current = true
AND staged.merge_action = 'UPDATE'
WHEN MATCHED THEN
UPDATE SET
target.is_current = false,
target.end_date = staged.effective_date
WHEN NOT MATCHED AND staged.merge_action IN ('INSERT', 'UPDATE') THEN
INSERT (worker_id, first_name, last_name, department,
cost_center, employment_status, row_hash,
effective_date, end_date, is_current)
VALUES (staged.worker_id, staged.first_name, staged.last_name,
staged.department, staged.cost_center, staged.employment_status,
staged.row_hash, staged.effective_date, staged.end_date, true)
""")

Stale Snapshot Cache: The Second Bug We Found

While diagnosing the MERGE issue, we discovered a second related problem: after the MERGE ran successfully, a subsequent read of the Iceberg table was returning duplicate active records — one from before the MERGE and one from after.

The cause: Iceberg snapshot caching in Glue’s Spark session. The Glue job had read the Iceberg table earlier in the same session, and that snapshot was cached. The MERGE wrote a new snapshot, but the read was still returning the old one.

Fix — force snapshot refresh before any post-MERGE read:

# After MERGE, always refresh the table snapshot
spark.sql("REFRESH TABLE glue_catalog.silver.dim_workers")
# Then read
result_df = spark.read.format("iceberg") \
.load("glue_catalog.silver.dim_workers")

Or, use the snapshot ID explicitly:

# Get latest snapshot ID
latest_snapshot = spark.sql("""
SELECT snapshot_id FROM glue_catalog.silver.dim_workers.snapshots
ORDER BY committed_at DESC LIMIT 1
""").collect()[0][0]
# Read at that snapshot
result_df = spark.read.format("iceberg") \
.option("snapshot-id", str(latest_snapshot)) \
.load("glue_catalog.silver.dim_workers")

Lessons and Prevention Checklist

After debugging both issues, here is what we added to our Glue job review checklist for every new Silver layer job:

Hash generation:

Use sha2(…, 256) or md5() for row hashing in SCD logic — never xxhash64() when the hash participates in a MERGE join key

If xxhash64 is needed for performance, add explicit row_number() deduplication before MERGE regardless of source cleanliness

MERGE safety:

Run source_df.groupBy(natural_key).count().filter(“count > 1”) before every MERGE in development — make it a unit test

Always use source_df.explain() when a new MERGE job is introduced to inspect the physical plan for non-deterministic expressions

Snapshot management:

Always REFRESH TABLE before reading an Iceberg table that was written to in the same Glue session

In multi-step jobs (MERGE → export → validate), pin reads to explicit snapshot IDs for reproducibility

Monitoring:

Set a CloudWatch alarm on glue.driver.aggregate.numFailedTasks > 0 to catch Spark task failures before the full job fails

Log the Iceberg snapshot ID after every MERGE to S3 as a lightweight audit trail

Summary

About the Author

Henil Patel is a Data Engineer building a production medallion lakehouse on AWS at Schellman, processing 50M+ records/month across Workday, HubSpot, QuickBase, and Kantata sources using Apache Iceberg, AWS Glue, MWAA, Redshift, and Terraform. He holds AWS Certified Data Engineer, GCP Professional Data Engineer, and Azure DP-203 certifications, and is pursuing an M.S. in Computer Science at Illinois Institute of Technology.

Connect on [LinkedIn]

Tags: Apache Iceberg, AWS Glue, PySpark, SCD Type 2, Data Engineering, Lakehouse, AWS, Spark, Delta Lake