Time travel and versioning in Apache Spark

Introduction

Time travel and versioning let you see and use past versions of your data easily. This helps fix mistakes or compare changes over time.

You want to check what your data looked like yesterday.

You accidentally deleted some data and want to restore it.

You want to compare sales data from last month to this month.

You need to audit changes made to your data over time.

You want to run reports on historical data snapshots.

Syntax

Apache Spark

spark.read.format("delta").option("versionAsOf", version_number).load(path)
spark.read.format("delta").option("timestampAsOf", "yyyy-MM-dd HH:mm:ss").load(path)

Use versionAsOf to load data from a specific version number.

Use timestampAsOf to load data as it was at a specific time.

Examples

Load the data as it was in version 3.

Apache Spark

df = spark.read.format("delta").option("versionAsOf", 3).load("/data/delta/events")

Load the data as it was on April 1, 2024 at 10 AM.

Apache Spark

df = spark.read.format("delta").option("timestampAsOf", "2024-04-01 10:00:00").load("/data/delta/events")

Sample Program

This program creates a Delta table with two versions of data. It then reads and shows the current data and the data from the first version using time travel.

Apache Spark

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

spark = SparkSession.builder.appName("TimeTravelExample").getOrCreate()

# Path to Delta table
path = "/tmp/delta/events"

# Create initial data
data_v1 = [(1, "login"), (2, "logout")]
df_v1 = spark.createDataFrame(data_v1, ["user_id", "action"])
df_v1.write.format("delta").mode("overwrite").save(path)

# Update data - add a new event
data_v2 = [(1, "login"), (2, "logout"), (3, "purchase")]
df_v2 = spark.createDataFrame(data_v2, ["user_id", "action"])
df_v2.write.format("delta").mode("overwrite").save(path)

# Read current data
current_df = spark.read.format("delta").load(path)
print("Current data:")
current_df.show()

# Read data from version 1 (first save)
version1_df = spark.read.format("delta").option("versionAsOf", 1).load(path)
print("Data at version 1:")
version1_df.show()

spark.stop()

Important Notes

Time travel works only with Delta tables, not regular Parquet or CSV files.

Each save or write creates a new version automatically.

Older versions may be cleaned up if retention policies are applied, so keep backups if needed.

Summary

Time travel lets you access past versions of your Delta data easily.

Use versionAsOf or timestampAsOf options to load old data.

This helps with data recovery, auditing, and comparing changes over time.

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