Merge (upsert) operations in Apache Spark - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

What is the output of this merge operation?

Given two Spark DataFrames df_target and df_updates, what will be the content of df_target after the merge?

Apache Spark

from pyspark.sql import SparkSession
from delta.tables import DeltaTable

spark = SparkSession.builder.appName('merge_test').getOrCreate()

# Initial target data
data_target = [(1, 'Alice', 100), (2, 'Bob', 200)]
df_target = spark.createDataFrame(data_target, ['id', 'name', 'score'])
df_target.write.format('delta').mode('overwrite').save('/tmp/delta/target')

delta_target = DeltaTable.forPath(spark, '/tmp/delta/target')

# Updates data
data_updates = [(2, 'Bob', 250), (3, 'Charlie', 300)]
df_updates = spark.createDataFrame(data_updates, ['id', 'name', 'score'])

# Perform merge (upsert)
delta_target.alias('t').merge(
    df_updates.alias('u'),
    't.id = u.id'
).whenMatchedUpdate(set={
    'score': 'u.score'
}).whenNotMatchedInsert(values={
    'id': 'u.id',
    'name': 'u.name',
    'score': 'u.score'
}).execute()

# Read final data
result = spark.read.format('delta').load('/tmp/delta/target').orderBy('id').collect()
output = [(row['id'], row['name'], row['score']) for row in result]
print(output)

A [(1, 'Alice', 100), (2, 'Bob', 200), (3, 'Charlie', 300)]

B [(1, 'Alice', 100), (2, 'Bob', 250), (3, 'Charlie', 300)]

C [(2, 'Bob', 250), (3, 'Charlie', 300)]

D [(1, 'Alice', 100), (2, 'Bob', 250)]

💡 Hint

Think about how matched rows are updated and unmatched rows are inserted.

The merge updates the score for id=2 from 200 to 250 and inserts the new row with id=3. The row with id=1 remains unchanged.

❓ data_output

intermediate

How many rows will the target DataFrame have after this merge?

Consider the following merge operation between df_target and df_updates. How many rows will df_target contain after the merge?

Apache Spark

from pyspark.sql import SparkSession
from delta.tables import DeltaTable

spark = SparkSession.builder.appName('merge_count').getOrCreate()

# Target data
data_target = [(1, 'Anna', 50), (2, 'Ben', 60), (3, 'Cara', 70)]
df_target = spark.createDataFrame(data_target, ['id', 'name', 'score'])
df_target.write.format('delta').mode('overwrite').save('/tmp/delta/target_count')

delta_target = DeltaTable.forPath(spark, '/tmp/delta/target_count')

# Updates data
data_updates = [(2, 'Ben', 65), (4, 'Derek', 80)]
df_updates = spark.createDataFrame(data_updates, ['id', 'name', 'score'])

# Merge operation
delta_target.alias('t').merge(
    df_updates.alias('u'),
    't.id = u.id'
).whenMatchedUpdate(set={'score': 'u.score'}).whenNotMatchedInsert(values={'id': 'u.id', 'name': 'u.name', 'score': 'u.score'}).execute()

# Count rows after merge
count = spark.read.format('delta').load('/tmp/delta/target_count').count()
print(count)

A 4

B 3

C 5

D 2

💡 Hint

Rows with matching ids update, new ids insert.

The original 3 rows remain, one row is updated, and one new row is inserted, totaling 4 rows.

🔧 Debug

advanced

What error does this merge code raise?

This code attempts a merge operation but fails. What error will it raise?

Apache Spark

from pyspark.sql import SparkSession
from delta.tables import DeltaTable

spark = SparkSession.builder.appName('merge_error').getOrCreate()

data_target = [(1, 'X', 10)]
df_target = spark.createDataFrame(data_target, ['id', 'name', 'score'])
df_target.write.format('delta').mode('overwrite').save('/tmp/delta/error')

delta_target = DeltaTable.forPath(spark, '/tmp/delta/error')

data_updates = [(1, 'X', 20)]
df_updates = spark.createDataFrame(data_updates, ['id', 'name', 'score'])

# Incorrect merge condition (missing alias)
delta_target.merge(
    df_updates,
    'id = id'
).whenMatchedUpdate(set={'score': 'df_updates.score'}).execute()

A AnalysisException: Reference 'id' is ambiguous, could be: id, id.

B TypeError: merge() missing 1 required positional argument

C SyntaxError: invalid syntax in merge condition

D No error, merge runs successfully

💡 Hint

Check how the merge condition references columns without aliases.

The merge condition 'id = id' is ambiguous because both DataFrames have a column named 'id'. Spark cannot tell which 'id' is which without aliases.

🚀 Application

advanced

Which option correctly performs an upsert with multiple column updates?

You want to merge df_updates into df_target updating both score and name columns when matched, and insert all columns when not matched. Which code snippet does this correctly?

A

delta_target.alias('t').merge(df_updates.alias('u'), 't.id = u.id')
  .whenMatchedUpdate(set={'score': 'u.score'})
  .whenNotMatchedInsert(values={'id': 'u.id', 'name': 'u.name'})
  .execute()

B

delta_target.merge(df_updates, 't.id = u.id')
  .whenMatchedUpdate(set={'score': 'u.score', 'name': 'u.name'})
  .whenNotMatchedInsertAll()
  .execute()

C

delta_target.alias('t').merge(df_updates.alias('u'), 't.id = u.id')
  .whenMatchedUpdate(set={'score': 'u.score', 'name': 'u.name'})
  .whenNotMatchedInsert(values={'id': 'u.id', 'score': 'u.score'})
  .execute()

D

delta_target.alias('t').merge(df_updates.alias('u'), 't.id = u.id')
  .whenMatchedUpdate(set={'score': 'u.score', 'name': 'u.name'})
  .whenNotMatchedInsertAll()
  .execute()

💡 Hint

Check for correct alias usage and full insert of all columns.

Option D uses aliases correctly, updates both columns on match, and inserts all columns on no match using whenNotMatchedInsertAll().

🧠 Conceptual

expert

What is the main advantage of using merge (upsert) operations in Delta Lake?

Select the best explanation for why merge (upsert) operations are important in Delta Lake workflows.

A They speed up data loading by skipping all validation and schema checks.

B They automatically partition data without user configuration.

C They allow combining insert, update, and delete operations in a single atomic transaction, ensuring data consistency.

D They convert Delta tables into Parquet files for faster queries.

💡 Hint

Think about data consistency and atomicity in big data pipelines.

Merge operations allow multiple data changes (insert, update, delete) in one atomic step, which keeps data consistent and avoids partial updates.

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