ACID transactions on data lakes in Apache Spark - Mini Project: Build & Apply

ACID Transactions on Data Lakes with Apache Spark

📖 Scenario: You work at a company that stores sales data in a data lake. You want to make sure that updates to the sales data are safe and reliable. This means if something goes wrong during an update, the data should not get corrupted or lost. This is called ACID transactions.Apache Spark with Delta Lake helps you do this by allowing you to write and update data with ACID guarantees.

🎯 Goal: Build a simple Spark program that creates a Delta Lake table, writes initial sales data, updates some records safely using ACID transactions, and then reads the final data.

📋 What You'll Learn

Create a Spark session with Delta Lake support

Create initial sales data as a DataFrame

Write the data to a Delta Lake table

Update the Delta table using ACID transaction

Read and display the final data

💡 Why This Matters

🌍 Real World

Companies use data lakes to store large amounts of data. ACID transactions ensure data stays correct and consistent even when many users update data at the same time.

💼 Career

Data engineers and data scientists use Delta Lake with Apache Spark to build reliable data pipelines and analytics platforms that handle updates safely.

1

Create initial sales data as a Spark DataFrame

Create a Spark DataFrame called sales_df with these exact rows: (1, '2024-01-01', 100), (2, '2024-01-02', 150), and (3, '2024-01-03', 200). The columns should be sale_id, date, and amount.

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType

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

# Create sales_df DataFrame with sale_id, date, amount columns and 3 rows
# Your code here

Hint

Use spark.createDataFrame(data, schema) with a list of tuples for data and a schema defining the columns.

2

Set Delta Lake table path

Create a string variable called delta_table_path and set it to "/tmp/delta_sales". This will be the location where the Delta table is saved.

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType

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

schema = StructType([
    StructField('sale_id', IntegerType(), False),
    StructField('date', StringType(), False),
    StructField('amount', IntegerType(), False)
])
data = [
    (1, '2024-01-01', 100),
    (2, '2024-01-02', 150),
    (3, '2024-01-03', 200)
]
sales_df = spark.createDataFrame(data, schema)

# Create delta_table_path variable with value '/tmp/delta_sales'
# Your code here

Hint

Just assign the string '/tmp/delta_sales' to the variable delta_table_path.

3

Write initial data to Delta Lake table and update a record

Write the sales_df DataFrame to the Delta Lake table at delta_table_path using format('delta') and mode('overwrite'). Then create a new DataFrame called update_df with one row: (2, '2024-01-02', 175) to update the amount for sale_id 2. Use Delta Lake's merge to update the record in the Delta table where sale_id matches.

Apache Spark

from delta.tables import DeltaTable

# Write sales_df to delta_table_path with format 'delta' and mode 'overwrite'
sales_df.write.format('delta').mode('overwrite').save(delta_table_path)

# Create update_df DataFrame with (2, '2024-01-02', 175)
# Your code here

# Load DeltaTable from delta_table_path
# Your code here

# Use merge to update amount where sale_id matches
# Your code here

Hint

Use sales_df.write.format('delta').mode('overwrite').save(delta_table_path) to write initial data. Then create update_df with the updated row. Use DeltaTable.forPath to load the table and merge to update the matching sale_id.

4

Read and display the final Delta Lake table data

Read the Delta Lake table from delta_table_path into a DataFrame called final_df using format('delta'). Then print the contents of final_df using show().

Apache Spark

from pyspark.sql import SparkSession

# Read Delta table from delta_table_path into final_df
# Your code here

# Show the contents of final_df
# Your code here

Hint

Use spark.read.format('delta').load(delta_table_path) to read the table, then final_df.show() to display.

Practice

1. What does ACID stand for in the context of data lakes?

easy

A. Aggregation, Calculation, Insertion, Deletion

B. Accuracy, Clarity, Integrity, Data

C. Access, Control, Indexing, Distribution

D. Atomicity, Consistency, Isolation, Durability

Solution

1. Step 1: Understand ACID properties

   ACID is a set of properties that guarantee reliable processing of database transactions.

2. Step 2: Match each letter to its meaning

   Atomicity means all or nothing, Consistency means data stays correct, Isolation means transactions don't interfere, Durability means changes are permanent.

3. Final Answer:

   Atomicity, Consistency, Isolation, Durability -> Option D

4. Quick Check:

   ACID = Atomicity, Consistency, Isolation, Durability [OK]

Hint: Remember ACID as key database safety rules [OK]

Common Mistakes:

• Confusing ACID with unrelated terms
• Mixing up the letters' meanings
• Thinking ACID is only for SQL databases

2. Which syntax correctly creates a Delta table with ACID support in Apache Spark?

easy

A. spark.sql("CREATE TABLE delta_table TYPE delta LOCATION '/delta/path'")

B. spark.sql("CREATE DELTA TABLE delta_table AS '/delta/path'")

C. spark.sql("CREATE TABLE delta_table USING delta LOCATION '/delta/path'")

D. spark.sql("MAKE TABLE delta_table USING delta LOCATION '/delta/path'")

Solution

1. Step 1: Recall Delta Lake table creation syntax

   The correct syntax uses CREATE TABLE with USING delta and LOCATION to specify path.

2. Step 2: Identify correct syntax among options

   spark.sql("CREATE TABLE delta_table USING delta LOCATION '/delta/path'") matches the correct syntax exactly; others have wrong keywords or order.

3. Final Answer:

   spark.sql("CREATE TABLE delta_table USING delta LOCATION '/delta/path'") -> Option C

4. Quick Check:

   CREATE TABLE ... USING delta ... = spark.sql("CREATE TABLE delta_table USING delta LOCATION '/delta/path'") [OK]

Hint: Look for 'CREATE TABLE ... USING delta' syntax [OK]

Common Mistakes:

• Using 'MAKE TABLE' instead of 'CREATE TABLE'
• Wrong keyword order or missing USING
• Confusing AS with LOCATION

3. Given the following code snippet, what will be the output count after the update?

from delta.tables import DeltaTable

# Assume delta_table is a DeltaTable object
# Initial data has 3 rows with column 'value' = [10, 20, 30]
delta_table.update("value = 20", {"value": "value + 5"})
count = delta_table.toDF().count()
print(count)

medium

A. 4

B. 3

C. 2

D. 0

Solution

1. Step 1: Understand the update operation

   The update changes rows where value = 20 by adding 5, so that row's value becomes 25.

2. Step 2: Check if row count changes

   Update modifies data in place; it does not add or remove rows, so count remains 3.

3. Final Answer:

   3 -> Option B

4. Quick Check:

   Update changes values, not row count = 3 [OK]

Hint: Update changes data but not number of rows [OK]

Common Mistakes:

• Thinking update adds new rows
• Confusing update with insert
• Assuming count changes after update

4. Identify the error in this Delta Lake transaction code:

delta_table.update("value > 10", {"value": "value + 1"})
delta_table.delete("value < 5")
delta_table.commit()

medium

A. DeltaTable has no commit() method; transactions auto-commit

B. Update condition syntax is incorrect

C. Delete condition should be a function, not a string

D. Missing startTransaction() before update

Solution

1. Step 1: Check DeltaTable API methods

   DeltaTable supports update and delete, but does not have a commit() method; changes auto-commit.

2. Step 2: Verify other method usages

   Update and delete accept string conditions correctly; no need for explicit transaction start.

3. Final Answer:

   DeltaTable has no commit() method; transactions auto-commit -> Option A

4. Quick Check:

   No commit() in DeltaTable API = DeltaTable has no commit() method; transactions auto-commit [OK]

Hint: Delta Lake auto-commits; no commit() needed [OK]

Common Mistakes:

• Trying to call commit() on DeltaTable
• Misunderstanding condition parameter types
• Expecting manual transaction control

5. You want to ensure multiple users can safely update a Delta Lake table without conflicts. Which feature helps achieve this?

hard

A. Delta Lake's optimistic concurrency control with ACID transactions

B. Using Spark's cache() to lock the table

C. Manually locking files in the storage system

D. Running updates sequentially with sleep delays

Solution

1. Step 1: Understand concurrency in Delta Lake

   Delta Lake uses optimistic concurrency control to handle multiple users updating data safely.

2. Step 2: Evaluate other options

   Cache() only speeds reads, manual file locks are error-prone, and sleep delays are unreliable.

3. Final Answer:

   Delta Lake's optimistic concurrency control with ACID transactions -> Option A

4. Quick Check:

   Optimistic concurrency control = safe multi-user updates [OK]

Hint: Trust Delta Lake's built-in concurrency control [OK]

Common Mistakes:

• Thinking cache() locks data
• Trying manual file locks
• Using delays instead of transaction control