14. Merge & Upsert Operations in PySpark
DfMerge (Upsert)
A merge operation combines INSERT, UPDATE, and DELETE operations in a single atomic transaction based on a match condition. It is the foundation of Change Data Capture (CDC) and slowly changing dimension (SCD) patterns in Delta Lake.
DfChange Data Capture (CDC)
CDC is a technique that captures and tracks changes (inserts, updates, deletes) made to data in a source system, enabling downstream systems to apply the same changes to maintain synchronized copies.
Merge File Pruning Efficiency
Here,
- =Fraction of files skipped via data skipping (target > 0.9)
- =Number of Parquet files that need to be read
- =Total number of Parquet files in the target table
Delta Lake merge uses data skipping via file-level min/max statistics to prune irrelevant files. To maximize pruning, ensure your merge key is clustered in the target table using ZORDER BY.
For high-frequency upserts, use MERGE with OPTIMIZE ZORDER BY on the merge key to maintain file-level statistics. Run VACUUM periodically to remove old files and prevent metadata bloat.
ThMerge Atomicity
Theorem: Delta Lake merge is atomic at the file level — it reads existing files, applies transformations, writes new files, and atomically updates the Delta log in a single commit. If any step fails, the transaction is rolled back and the table remains in its previous consistent state.
- Merge = atomic INSERT + UPDATE + DELETE based on match condition
- CDC captures source changes for downstream synchronization
- File pruning via min/max statistics is critical for merge performance
- Use
OPTIMIZE ZORDER BYon merge keys; useVACUUMto clean old files
🏗️ Merge Operation Architecture
┌─────────────────────────────────────────────────────────────────────────┐
│ MERGE OPERATION ARCHITECTURE │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ SOURCE DATA │ │
│ │ │ │
│ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │
│ │ │ New Records │ │ Updated │ │ Deleted │ │ │
│ │ │ (INSERT) │ │ Records │ │ Records │ │ │
│ │ │ │ │ (UPDATE) │ │ (DELETE) │ │ │
│ │ └──────────────┘ └──────────────┘ └──────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ MERGE ENGINE │ │
│ │ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ 1. MATCH DETECTION │ │ │
│ │ │ • Join condition evaluation │ │ │
│ │ │ • Duplicate detection │ │ │
│ │ │ • Conflict resolution │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ 2. ACTION DETERMINATION │ │ │
│ │ │ • INSERT (new records) │ │ │
│ │ │ • UPDATE (matched records) │ │ │
│ │ │ • DELETE (matched records) │ │ │
│ │ │ • MATCHED BY SOURCE (for CDC) │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ 3. TRANSACTION EXECUTION │ │ │
│ │ │ • Atomic write │ │ │
│ │ │ • Version management │ │ │
│ │ │ • Conflict resolution │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ TARGET TABLE (DELTA) │ │
│ │ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ Version N: [old data] │ │ │
│ │ │ Version N+1: [merged data] (after merge operation) │ │ │
│ │ │ Version N+2: [further updates] │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────┘
🔄 CDC Pattern Diagram
┌─────────────────────────────────────────────────────────────────────────┐
│ CHANGE DATA CAPTURE (CDC) PATTERNS │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ CDC SOURCE (Database) │ │
│ │ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ Operation Log │ │ │
│ │ │ ┌─────────┬─────────┬─────────┬─────────┬─────────┐ │ │ │
│ │ │ │ INSERT │ UPDATE │ DELETE │ INSERT │ UPDATE │ │ │ │
│ │ │ │ id=1 │ id=2 │ id=3 │ id=4 │ id=5 │ │ │ │
│ │ │ └─────────┴─────────┴─────────┴─────────┴─────────┘ │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ CDC PROCESSING │ │
│ │ │ │
│ │ ┌─────────────────┐ ┌─────────────────┐ │ │
│ │ │ Extract │───▶│ Transform │ │ │
│ │ │ • Log parsing │ │ • Schema map │ │ │
│ │ │ • Filtering │ │ • Enrichment │ │ │
│ │ │ • Validation │ │ • Dedup │ │ │
│ │ └─────────────────┘ └─────────────────┘ │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌─────────────────┐ ┌─────────────────┐ │ │
│ │ │ Load │◀───│ Merge │ │ │
│ │ │ • Delta write │ │ • Match │ │ │
│ │ │ • Versioning │ │ • Apply │ │ │
│ │ │ • Cleanup │ │ • Resolve │ │ │
│ │ └─────────────────┘ └─────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ CDC DATA FLOW │ │
│ │ │ │
│ │ Source ──▶ Extract ──▶ Transform ──▶ Merge ──▶ Target │ │
│ │ │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │ DB Log │ │ Stream │ │ Buffer │ │ Delta │ │ │
│ │ │ (WAL) │ │ (Kafka) │ │ (Memory)│ │ Table │ │ │
│ │ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │ │
│ │ │ │
│ │ Latency: seconds ──▶ minutes ──▶ seconds ──▶ seconds │ │
│ └─────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────┘
⚡ Upsert Strategies Diagram
┌─────────────────────────────────────────────────────────────────────────┐
│ UPSERT STRATEGIES COMPARISON │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ STRATEGY 1: MERGE (Default) │ │
│ │ │ │
│ │ Source ──▶ [Join] ──▶ [Match] ──▶ [Apply Actions] │ │
│ │ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ • Matches on key │ │ │
│ │ │ • Applies INSERT/UPDATE/DELETE based on conditions │ │ │
│ │ │ • Handles all CDC operations │ │ │
│ │ │ • Performance: Moderate (requires join) │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ STRATEGY 2: INSERT OVERWRITE │ │
│ │ │ │
│ │ Source ──▶ [Replace Partition] ──▶ [Write] │ │
│ │ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ • Replaces entire partition │ │ │
│ │ │ • No partial updates │ │ │
│ │ │ • Simple but destructive │ │ │
│ │ │ • Performance: Fast (no join required) │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ STRATEGY 3: APPEND + DEDUP │ │
│ │ │ │
│ │ Source ──▶ [Append] ──▶ [Deduplicate] ──▶ [Compact] │ │
│ │ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ • Appends all records │ │ │
│ │ │ • Deduplicates on read │ │ │
│ │ │ • Requires dedup logic │ │ │
│ │ │ • Performance: Fast write, slow read │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────────────────┐ │
│ │ PERFORMANCE COMPARISON │ │
│ │ │ │
│ │ ┌─────────────────────────────────────────────────────────┐ │ │
│ │ │ Strategy │ Write │ Read │ Storage │ Complexity│ │ │
│ │ │──────────────────┼────────┼────────┼─────────┼───────────│ │ │
│ │ │ MERGE │ Medium │ Fast │ Low │ High │ │ │
│ │ │ INSERT OVERWRITE│ Fast │ Fast │ Medium │ Low │ │ │
│ │ │ APPEND + DEDUP │ Fast │ Slow │ High │ Medium │ │ │
│ │ └─────────────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────┘
📚 Detailed Explanation
Merge and upsert operations are fundamental patterns in modern data lakes, enabling efficient updates, inserts, and deletes on large datasets. In PySpark, these operations are primarily implemented through Delta Lake's merge functionality, which provides ACID transactions, versioning, and efficient change data capture (CDC) processing.
The merge operation is the most flexible and powerful upsert strategy. It allows you to specify match conditions and define separate actions for matched and unmatched records. This makes it ideal for CDC scenarios where you need to handle inserts, updates, and deletes in a single operation. The merge operation uses a join to match source records with target records, then applies the specified actions based on the match results.
CDC patterns represent a critical use case for merge operations. In CDC, changes from a source system (typically a database) are captured and applied to a target system (usually a data lake). The merge operation handles the complexity of applying these changes efficiently, including handling late-arriving data, resolving conflicts, and maintaining data consistency.
The choice of upsert strategy depends on several factors: data volume, update patterns, latency requirements, and complexity constraints. The merge operation is the most flexible but has higher overhead due to the join. Insert overwrite is faster but doesn't support partial updates. Append with dedup is the fastest for writes but has higher storage costs and slower reads.
Performance optimization for merge operations involves several techniques: partitioning the target table to reduce the join scope, using efficient join strategies, optimizing the merge conditions, and managing file sizes through compaction. The target table's partition strategy is particularly important, as it directly affects the join performance.
Conflict resolution is a critical aspect of merge operations. When multiple writers attempt to merge into the same table simultaneously, conflicts can occur. Delta Lake provides optimistic concurrency control to handle these conflicts, automatically retrying operations when conflicts are detected. Understanding and configuring conflict resolution is essential for production systems.
Data quality considerations include handling schema evolution during merge operations, validating data types and constraints, and ensuring referential integrity. Delta Lake's schema enforcement and evolution capabilities help maintain data quality during merge operations.
Monitoring and observability are important for maintaining merge operation health. Key metrics include merge duration, number of records processed, file compaction rates, and conflict frequencies. These metrics help identify performance bottlenecks and optimize the merge strategy.
Advanced techniques include conditional merges, where different actions are applied based on complex conditions, and merge with aggregation, where source records are aggregated before merging. These techniques enable sophisticated data processing patterns while maintaining the benefits of ACID transactions.
📊 Key Concepts Table
| Concept | Description | Use Case |
|---|---|---|
| Merge Operation | Join-based upsert with conditional actions | CDC, complex updates |
| CDC (Change Data Capture) | Capturing and applying database changes | Data synchronization |
| Upsert | Insert or update based on existence | Real-time data ingestion |
| Partition Pruning | Reducing join scope through partitioning | Performance optimization |
| Optimistic Concurrency | Conflict resolution for concurrent writes | Multi-writer scenarios |
| Schema Evolution | Handling schema changes during merge | Evolving data structures |
| File Compaction | Merging small files for better performance | Storage optimization |
| Time Travel | Accessing historical versions | Auditing, debugging |
💻 Code Examples
Basic Merge Operation
from pyspark.sql import SparkSession
from delta.tables import DeltaTable
from pyspark.sql.functions import *
spark = SparkSession.builder \
.appName("MergeOperation") \
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
.getOrCreate()
# Read target Delta table
target_table = DeltaTable.forPath(spark, "/path/to/target/table")
# Read source data (e.g., from Kafka, files, etc.)
source_df = spark.read \
.format("json") \
.load("/path/to/source/data")
# Perform merge operation
target_table.alias("target").merge(
source_df.alias("source"),
"target.id = source.id" # Match condition
).whenMatchedUpdate(
condition="target.last_updated < source.last_updated", # Only update if source is newer
set={
"name": "source.name",
"email": "source.email",
"last_updated": "source.last_updated"
}
).whenNotMatchedInsert(
values={
"id": "source.id",
"name": "source.name",
"email": "source.email",
"last_updated": "source.last_updated",
"created_at": current_timestamp()
}
).execute()
print("Merge operation completed successfully")
Advanced Merge with Delete
from pyspark.sql import SparkSession
from delta.tables import DeltaTable
from pyspark.sql.functions import *
spark = SparkSession.builder \
.appName("AdvancedMerge") \
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
.getOrCreate()
# Read target Delta table
target_table = DeltaTable.forPath(spark, "/path/to/target/table")
# Read source data with operation column
source_df = spark.read \
.format("delta") \
.load("/path/to/cdc/source") \
.withColumn("operation", col("op")) # o=insert, u=update, d=delete
# Perform merge with delete operation
target_table.alias("target").merge(
source_df.alias("source"),
"target.id = source.id"
).whenMatchedUpdate(
condition="source.operation IN ('o', 'u')", # Insert or Update
set={
"name": "source.name",
"email": "source.email",
"status": "source.status",
"last_updated": "source.timestamp"
}
).whenMatchedDelete(
condition="source.operation = 'd'" # Delete operation
).whenNotMatchedInsert(
condition="source.operation IN ('o', 'u')", # Only insert for new records
values={
"id": "source.id",
"name": "source.name",
"email": "source.email",
"status": "source.status",
"last_updated": "source.timestamp",
"created_at": "source.timestamp"
}
).execute()
print("CDC merge operation completed successfully")
Merge with Partition Pruning
from pyspark.sql import SparkSession
from delta.tables import DeltaTable
from pyspark.sql.functions import *
spark = SparkSession.builder \
.appName("PartitionPruningMerge") \
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
.getOrCreate()
# Read target Delta table (partitioned by date)
target_table = DeltaTable.forPath(spark, "/path/to/partitioned/table")
# Read source data for specific date
source_df = spark.read \
.format("json") \
.load("/path/to/source/data") \
.filter(col("event_date") == "2024-01-15") # Filter to specific partition
# Perform merge with partition pruning
target_table.alias("target").merge(
source_df.alias("source"),
"target.id = source.id AND target.event_date = source.event_date" # Include partition key
).whenMatchedUpdate(
set={
"value": "source.value",
"updated_at": current_timestamp()
}
).whenNotMatchedInsert(
values={
"id": "source.id",
"event_date": "source.event_date",
"value": "source.value",
"created_at": current_timestamp(),
"updated_at": current_timestamp()
}
).execute()
print("Partition-pruned merge completed successfully")
Merge with Aggregation
from pyspark.sql import SparkSession
from delta.tables import DeltaTable
from pyspark.sql.functions import *
spark = SparkSession.builder \
.appName("MergeWithAggregation") \
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
.getOrCreate()
# Read target Delta table
target_table = DeltaTable.forPath(spark, "/path/to/target/table")
# Read source data and aggregate
source_df = spark.read \
.format("json") \
.load("/path/to/source/data") \
.groupBy("user_id", "event_date") \
.agg(
sum("amount").alias("total_amount"),
count("*").alias("event_count"),
max("timestamp").alias("last_event_time")
)
# Perform merge with aggregated data
target_table.alias("target").merge(
source_df.alias("source"),
"target.user_id = source.user_id AND target.event_date = source.event_date"
).whenMatchedUpdate(
set={
"total_amount": "target.total_amount + source.total_amount",
"event_count": "target.event_count + source.event_count",
"last_event_time": "GREATEST(target.last_event_time, source.last_event_time)",
"updated_at": current_timestamp()
}
).whenNotMatchedInsert(
values={
"user_id": "source.user_id",
"event_date": "source.event_date",
"total_amount": "source.total_amount",
"event_count": "source.event_count",
"last_event_time": "source.last_event_time",
"created_at": current_timestamp(),
"updated_at": current_timestamp()
}
).execute()
print("Merge with aggregation completed successfully")
📈 Performance Metrics
| Metric | Optimized | Typical | Poor | Optimization |
|---|---|---|---|---|
| Merge Duration | < 30s | 30s-5min | > 5min | Partition pruning, file optimization |
| Records Processed/sec | > 100K | 10K-100K | < 10K | Parallelism, join optimization |
| File Compaction | < 100 files | 100-1000 files | > 1000 files | Regular compaction, optimize command |
| Conflict Rate | < 1% | 1-5% | > 5% | Concurrency control, partitioning |
| Storage Efficiency | > 80% | 60-80% | < 60% | Compaction, z-ordering |
🏆 Best Practices
- Partition strategically - Partition by high-cardinality columns used in join conditions
- Use partition pruning - Always include partition keys in merge conditions
- Optimize file sizes - Aim for 128MB-1GB files for optimal performance
- Regular compaction - Run OPTIMIZE command to merge small files
- Z-order indexing - Create z-order indexes on frequently filtered columns
- Monitor conflicts - Track and minimize merge conflicts
- Handle schema evolution - Use mergeSchema option for schema changes
- Test with realistic data - Validate performance with production-like volumes
- Use checkpointing - For streaming merges, configure appropriate checkpoint intervals
- Backup before major operations - Always backup data before large merge operations
🔗 Related Topics
- 11-structured-streaming.mdx: Streaming architecture for real-time merges
- 12-state-management.mdx: Stateful operations for merge patterns
- 13-window-operations.mdx: Windowed aggregations before merge
- 16-schema-evolution.mdx: Schema changes during merge operations
See Also
- Kafka Streams (kafka/03): Stream-table duality for upsert semantics
- Data Engineering Streaming (data-engineering/022): CDC merge patterns in streaming pipelines