Merge (upsert) operations in Apache Spark - Step-by-Step Execution

Concept Flow - Merge (upsert) operations

Start with Target Table

↓

Read Source Data

↓

Compare Keys

↓

Match

↓

Update Rows

↓

Write Back to Target

↓

End

The merge operation compares source data with target table keys, updates matching rows, inserts new rows, and writes the result back.

Execution Sample

Apache Spark

from delta.tables import DeltaTable

source_df = spark.createDataFrame([(1, 'A'), (3, 'C')], ['id', 'value'])
deltaTable = DeltaTable.forName(spark, 'target_table')

deltaTable.alias('t').merge(
    source_df.alias('s'), 't.id = s.id'
).whenMatchedUpdate(set = {'value': 's.value'})
.whenNotMatchedInsert(values = {'id': 's.id', 'value': 's.value'})
.execute()

This code merges source_df into target_table by updating matching ids and inserting new ones.

Execution Table

Step	Action	Source Row	Target Row	Condition (t.id = s.id)	Operation	Resulting Target Table
1	Compare source row (id=1) with target rows	(1, 'A')	(1, 'X')	True	Update value to 'A'	Row with id=1 updated to value 'A'
2	Compare source row (id=3) with target rows	(3, 'C')	No match	False	Insert new row (3, 'C')	Row with id=3 inserted
3	Write back changes	N/A	N/A	N/A	Commit updates and inserts	Target table now has updated and new rows

💡 All source rows processed; merge operation complete.

Variable Tracker

Variable	Start	After Step 1	After Step 2	Final
target_table	[ (1, 'X'), (2, 'B') ]	[ (1, 'A'), (2, 'B') ]	[ (1, 'A'), (2, 'B'), (3, 'C') ]	[ (1, 'A'), (2, 'B'), (3, 'C') ]
source_df	[ (1, 'A'), (3, 'C') ]	[ (1, 'A'), (3, 'C') ]	[ (1, 'A'), (3, 'C') ]	[ (1, 'A'), (3, 'C') ]

Key Moments - 3 Insights

Why does the row with id=1 get updated instead of inserted?

Because the condition 't.id = s.id' is True for id=1 (see execution_table step 1), so the merge updates the existing row instead of inserting.

What happens when a source row has no matching target row?

The merge inserts the source row as new (see execution_table step 2), because the condition is False and 'whenNotMatchedInsert' applies.

Does the target table change immediately after each row is processed?

No, changes are staged during merge and committed at the end (see execution_table step 3), so the final target table reflects all updates and inserts together.

Visual Quiz - 3 Questions

Look at the execution_table, what operation happens at step 2?

A Update existing row

B Insert new row

C Delete row

D No operation

💡 Hint

Check the 'Operation' column in execution_table row 2.

According to variable_tracker, what is the value of target_table after step 1?

A [ (1, 'A'), (2, 'B') ]

B [ (1, 'A'), (2, 'B'), (3, 'C') ]

C [ (1, 'X'), (2, 'B') ]

D [ (3, 'C'), (2, 'B') ]

💡 Hint

Look at the 'After Step 1' column for target_table in variable_tracker.

If source_df had a row with id=2, what would happen during merge?

A Row with id=2 would be inserted

B Row with id=2 would be deleted

C Row with id=2 would be updated

D No change for id=2

💡 Hint

Recall merge updates matching keys (see execution_table step 1).

Concept Snapshot

Merge (upsert) operations in Spark:
- Compare source and target on key condition
- Update rows where keys match
- Insert rows where no match
- Commit all changes atomically
- Use DeltaTable.merge() with whenMatchedUpdate and whenNotMatchedInsert

Full Transcript

Merge (upsert) operations combine source data with a target table by matching keys. For each source row, if the key matches a target row, the target row is updated. If no match is found, the source row is inserted as new. All changes are committed together at the end. This is done in Spark using DeltaTable.merge(), specifying update and insert actions. The execution table shows step-by-step how each source row is compared and merged, and the variable tracker shows how the target table changes after each step.

Practice

1. What is the main purpose of a merge (upsert) operation in Apache Spark with Delta Lake?

easy

A. To update existing rows and insert new rows in one step

B. To delete all rows from a table

C. To create a new Spark session

D. To read data from a CSV file

Solution

1. Step 1: Understand the merge operation concept

   Merge (upsert) combines update and insert actions to keep data synchronized.

2. Step 2: Identify the main use case in Spark with Delta Lake

   It updates existing rows and inserts new rows efficiently in one command.

3. Final Answer:

   To update existing rows and insert new rows in one step -> Option A

4. Quick Check:

   Merge = update + insert [OK]

Hint: Merge means update existing and insert new rows together [OK]

Common Mistakes:

• Confusing merge with delete operation
• Thinking merge only inserts data
• Assuming merge creates new tables

2. Which of the following is the correct syntax to start a merge operation in Spark SQL with Delta Lake?

easy

A. deltaTable.select(sourceDF).merge()

B. deltaTable.join(sourceDF).update()

C. deltaTable.insert(sourceDF).update()

D. deltaTable.merge(sourceDF, condition).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()

Solution

1. Step 1: Recall the merge syntax in Delta Lake

   The merge starts with deltaTable.merge(sourceDF, condition) followed by update and insert clauses.

2. Step 2: Match the full correct method chain

   Correct chain is whenMatchedUpdateAll(), whenNotMatchedInsertAll(), then execute().

3. Final Answer:

   deltaTable.merge(sourceDF, condition).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute() -> Option D

4. Quick Check:

   Correct merge syntax = deltaTable.merge(sourceDF, condition).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute() [OK]

Hint: Merge uses .merge().whenMatchedUpdateAll().whenNotMatchedInsertAll().execute() [OK]

Common Mistakes:

• Using join instead of merge
• Calling update before merge
• Missing execute() at the end

3. Given the following Spark code snippet, what will be the output count of rows in targetTable after the merge?

val sourceDF = spark.createDataFrame(Seq((1, "A"), (3, "C"))).toDF("id", "value")
val deltaTable = DeltaTable.forName("targetTable")
deltaTable.merge(sourceDF, "target.id = source.id")
  .whenMatchedUpdateExpr(Map("value" -> "source.value"))
  .whenNotMatchedInsertExpr(Map("id" -> "source.id", "value" -> "source.value"))
  .execute()
val count = spark.table("targetTable").count()

Assuming targetTable initially has rows with ids 1 and 2.

medium

A. 2

B. 3

C. 4

D. 1

Solution

1. Step 1: Analyze initial data in targetTable

   targetTable has ids 1 and 2 initially.

2. Step 2: Apply merge logic with sourceDF

   Source has ids 1 and 3. Id 1 matches and updates value. Id 3 does not match and inserts new row.

3. Step 3: Calculate final row count

   Rows after merge: ids 1 (updated), 2 (unchanged), 3 (inserted) -> total 3 rows.

4. Final Answer:

   3 -> Option B

5. Quick Check:

   Initial 2 + 1 inserted = 3 rows [OK]

Hint: Count = initial rows + new inserts after merge [OK]

Common Mistakes:

• Counting only updated rows
• Ignoring inserted rows
• Assuming rows are deleted

4. Identify the error in the following merge operation code snippet:

deltaTable.merge(sourceDF, "target.id = source.id")
  .whenMatchedUpdate(Map("value" -> "source.value"))
  .whenNotMatchedInsert(Map("id" -> "source.id", "value" -> "source.value"))
  .execute()

medium

A. The update and insert methods should be whenMatchedUpdateExpr and whenNotMatchedInsertExpr

B. The condition string should use backticks around column names

C. The execute() method is missing

D. The source DataFrame is not registered as a temp view

Solution

1. Step 1: Check method names for update and insert

   In Delta Lake merge, update and insert with expressions require whenMatchedUpdateExpr and whenNotMatchedInsertExpr.

2. Step 2: Verify other parts of the code

   Condition string and execute() are correct; sourceDF does not need temp view for merge.

3. Final Answer:

   The update and insert methods should be whenMatchedUpdateExpr and whenNotMatchedInsertExpr -> Option A

4. Quick Check:

   Use whenMatchedUpdateExpr and whenNotMatchedInsertExpr [OK]

Hint: Use whenMatchedUpdateExpr and whenNotMatchedInsertExpr for expression maps [OK]

Common Mistakes:

• Using whenMatchedUpdate instead of whenMatchedUpdateExpr
• Forgetting execute()
• Miswriting condition string syntax

5. You have a Delta table users with columns user_id and email. You receive a new DataFrame updatesDF with updated emails and new users. How do you write a merge operation to update emails for existing users and insert new users, ensuring no duplicates and preserving existing users without changes?

hard

A. Use filter to remove duplicates before inserting

B. Use join and then insert only new users

C. Use merge with condition on user_id, update email when matched, insert all columns when not matched

D. Overwrite the entire users table with updatesDF

Solution

1. Step 1: Define merge condition on user_id

   Match rows where users.user_id = updatesDF.user_id to identify existing users.

2. Step 2: Specify update and insert actions

   Update email for matched users, insert all columns for unmatched (new) users.

3. Step 3: Preserve existing users without changes

   Rows not matched remain unchanged automatically in merge.

4. Final Answer:

   Use merge with condition on user_id, update email when matched, insert all columns when not matched -> Option C

5. Quick Check:

   Merge = update matched + insert unmatched [OK]

Hint: Merge on user_id, update email if matched, insert if not [OK]

Common Mistakes:

• Overwriting entire table losing data
• Using join without update logic
• Not handling unmatched rows