Time travel and versioning in Apache Spark - Mini Project: Build & Apply

Time Travel and Versioning with Apache Spark

📖 Scenario: You work at a company that tracks sales data daily. Sometimes, data gets updated or corrected. You want to keep all versions of the data so you can see past states and changes over time.

🎯 Goal: Build a simple Apache Spark project that creates a versioned dataset, updates it, and then uses time travel to view previous versions.

📋 What You'll Learn

Create an initial DataFrame with sales data

Save the DataFrame as a Delta table to enable versioning

Update the Delta table with new data

Use time travel to query the previous version of the data

💡 Why This Matters

🌍 Real World

Companies often need to keep track of data changes over time to audit, debug, or analyze historical states.

💼 Career

Understanding Delta Lake versioning and time travel is important for data engineers and data scientists working with big data pipelines.

1

Create initial sales data 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 id, date, and amount.

Apache Spark

# Create the initial sales_df DataFrame with the specified rows and columns
# Your code here

Hint

Use spark.createDataFrame with a list of tuples and specify the column names.

2

Save DataFrame as Delta table for versioning

Save the sales_df DataFrame as a Delta table named sales_delta in the path /tmp/sales_delta. Use mode='overwrite' to create the table fresh.

Apache Spark

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('TimeTravelExample').getOrCreate()
sales_data = [(1, '2024-01-01', 100), (2, '2024-01-02', 150), (3, '2024-01-03', 200)]
sales_df = spark.createDataFrame(sales_data, ['id', 'date', 'amount'])

# Save sales_df as a Delta table at /tmp/sales_delta with overwrite mode
# Your code here

Hint

Use write.format('delta').mode('overwrite').save(path) to save the DataFrame as a Delta table.

3

Update the Delta table with new sales data

Create a new DataFrame called new_sales_df with these rows: (4, '2024-01-04', 250) and (2, '2024-01-02', 175). Then append this data to the Delta table at /tmp/sales_delta.

Apache Spark

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('TimeTravelExample').getOrCreate()
sales_data = [(1, '2024-01-01', 100), (2, '2024-01-02', 150), (3, '2024-01-03', 200)]
sales_df = spark.createDataFrame(sales_data, ['id', 'date', 'amount'])
sales_df.write.format('delta').mode('overwrite').save('/tmp/sales_delta')

# Create new_sales_df with updated rows and append to /tmp/sales_delta
# Your code here

Hint

Create the new DataFrame and use write.format('delta').mode('append').save(path) to add data.

4

Use time travel to query the previous version

Read the Delta table at /tmp/sales_delta using version 0 (the original data) into a DataFrame called version0_df. Then print the contents of version0_df.

Apache Spark

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('TimeTravelExample').getOrCreate()
sales_data = [(1, '2024-01-01', 100), (2, '2024-01-02', 150), (3, '2024-01-03', 200)]
sales_df = spark.createDataFrame(sales_data, ['id', 'date', 'amount'])
sales_df.write.format('delta').mode('overwrite').save('/tmp/sales_delta')

new_sales_data = [(4, '2024-01-04', 250), (2, '2024-01-02', 175)]
new_sales_df = spark.createDataFrame(new_sales_data, ['id', 'date', 'amount'])
new_sales_df.write.format('delta').mode('append').save('/tmp/sales_delta')

# Read version 0 of the Delta table into version0_df and print it
# Your code here

Hint

Use spark.read.format('delta').option('versionAsOf', 0).load(path) to read the first version.

Practice

1. What is the main purpose of time travel in Apache Spark's Delta Lake?

easy

A. To access previous versions of data easily

B. To speed up data processing

C. To create new tables automatically

D. To delete old data permanently

Solution

1. Step 1: Understand time travel concept

   Time travel allows users to look at past versions of data stored in Delta Lake.

2. Step 2: Identify main use

   This feature helps in recovering old data, auditing changes, and comparing versions.

3. Final Answer:

   To access previous versions of data easily -> Option A

4. Quick Check:

   Time travel = Access past data versions [OK]

Hint: Time travel means looking back at old data versions [OK]

Common Mistakes:

• Thinking time travel speeds up processing
• Confusing time travel with data deletion
• Assuming it creates new tables automatically

2. Which option is the correct way to load a Delta table at a specific version using Spark?

easy

A. spark.read.format('delta').option('version', 5).load('/data/table')

B. spark.read.format('delta').option('versionAsOf', 5).load('/data/table')

C. spark.read.format('delta').option('timestampAsOf', '5').load('/data/table')

D. spark.read.format('delta').option('versionAt', 5).load('/data/table')

Solution

1. Step 1: Recall correct option name for version

   The correct option to specify version is 'versionAsOf'.

2. Step 2: Check syntax correctness

   spark.read.format('delta').option('versionAsOf', 5).load('/data/table') uses 'versionAsOf' with a number, which is correct syntax.

3. Final Answer:

   spark.read.format('delta').option('versionAsOf', 5).load('/data/table') -> Option B

4. Quick Check:

   Use 'versionAsOf' to load specific version [OK]

Hint: Use 'versionAsOf' option to load specific version [OK]

Common Mistakes:

• Using 'version' instead of 'versionAsOf'
• Passing version as string in 'timestampAsOf'
• Using wrong option names like 'versionAt'

3. Given the code below, what will be the output count?

df = spark.read.format('delta').option('versionAsOf', 2).load('/data/events')
print(df.count())

Assuming version 2 of the table has 150 rows.

medium

A. Error due to wrong option

B. 0

C. Latest row count

D. 150

Solution

1. Step 1: Understand versionAsOf usage

   The option 'versionAsOf' loads the table exactly at version 2.

2. Step 2: Check row count at version 2

   Since version 2 has 150 rows, the count will be 150.

3. Final Answer:

   150 -> Option D

4. Quick Check:

   Count at version 2 = 150 [OK]

Hint: Count matches rows in specified version [OK]

Common Mistakes:

• Assuming count is zero for old versions
• Confusing latest count with versioned count
• Expecting error from correct syntax

4. Identify the error in the following code snippet for loading a Delta table at a timestamp:

df = spark.read.format('delta').option('timestampAsOf', 20230101).load('/data/sales')

medium

A. Load path is incorrect

B. timestampAsOf option does not exist

C. timestampAsOf value should be a string with date-time format

D. Missing .option('versionAsOf', )

Solution

1. Step 1: Check timestampAsOf option usage

   The 'timestampAsOf' option expects a string representing date and time, e.g., '2023-01-01T00:00:00'.

2. Step 2: Identify error in value type

   Here, 20230101 is an integer, not a string with proper format, causing error.

3. Final Answer:

   timestampAsOf value should be a string with date-time format -> Option C

4. Quick Check:

   timestampAsOf needs string date-time [OK]

Hint: Use string date-time format for timestampAsOf [OK]

Common Mistakes:

• Passing integer instead of string for timestamp
• Confusing versionAsOf with timestampAsOf
• Assuming option does not exist

5. You want to compare the number of rows between version 3 and version 5 of a Delta table located at '/data/logs'. Which code snippet correctly loads both versions for comparison?

hard

A. df_v3 = spark.read.format('delta').option('versionAsOf', 3).load('/data/logs')
df_v5 = spark.read.format('delta').option('versionAsOf', 5).load('/data/logs')

B. df_v3 = spark.read.format('delta').option('timestampAsOf', 3).load('/data/logs')
df_v5 = spark.read.format('delta').option('timestampAsOf', 5).load('/data/logs')

C. df_v3 = spark.read.format('delta').option('version', 3).load('/data/logs')
df_v5 = spark.read.format('delta').option('version', 5).load('/data/logs')

D. df_v3 = spark.read.format('delta').option('versionAsOf', '3').load('/data/logs')
df_v5 = spark.read.format('delta').option('versionAsOf', '5').load('/data/logs')

Solution

1. Step 1: Identify correct option for version number

   The option 'versionAsOf' expects an integer version number, not a string.

2. Step 2: Check code correctness for both versions

   df_v3 = spark.read.format('delta').option('versionAsOf', 3).load('/data/logs') df_v5 = spark.read.format('delta').option('versionAsOf', 5).load('/data/logs') uses 'versionAsOf' with integers 3 and 5, correctly loading both versions.

3. Final Answer:

   df_v3 = spark.read.format('delta').option('versionAsOf', 3).load('/data/logs') df_v5 = spark.read.format('delta').option('versionAsOf', 5).load('/data/logs') -> Option A

4. Quick Check:

   Use 'versionAsOf' with integers for version loading [OK]

Hint: Use versionAsOf with integers to load specific versions [OK]

Common Mistakes:

• Using timestampAsOf with integers
• Passing version numbers as strings
• Using non-existent 'version' option