Idempotent pipeline design in Apache Spark - Mini Project: Build & Apply

Idempotent Pipeline Design

📖 Scenario: You work as a data engineer building a data pipeline using Apache Spark. Your pipeline processes sales data daily. Sometimes the pipeline runs multiple times for the same day due to retries or errors. To avoid duplicate data, you want to design the pipeline so it is idempotent. This means running the pipeline multiple times with the same input will not create duplicate results.

🎯 Goal: Build a simple Spark pipeline that reads sales data, filters for a specific date, and writes the results to a target location. The pipeline should be idempotent by removing any existing data for that date before writing new data.

📋 What You'll Learn

Create a Spark DataFrame with sample sales data

Set a variable for the target processing date

Filter the DataFrame for the target date

Remove existing data for the target date before writing

Write the filtered data to a parquet file

Print the count of records written

💡 Why This Matters

🌍 Real World

Data pipelines often run multiple times due to failures or updates. Idempotent design prevents duplicate or inconsistent data in storage.

💼 Career

Data engineers and data scientists must ensure pipelines produce consistent results even if run repeatedly. This skill is essential for reliable data workflows.

1

Create sample sales data

Create a Spark DataFrame called sales_df with these exact rows: (1, '2024-06-01', 100.0), (2, '2024-06-02', 150.0), (3, '2024-06-01', 200.0). Use columns named 'id', 'date', and 'amount'.

Apache Spark

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

Hint

Use spark.createDataFrame with a list of tuples and a schema to create sales_df.

2

Set the target processing date

Create a variable called process_date and set it to the string '2024-06-01'.

Apache Spark

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

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

schema = StructType([
    StructField('id', IntegerType(), False),
    StructField('date', StringType(), False),
    StructField('amount', FloatType(), False)
])
data = [
    (1, '2024-06-01', 100.0),
    (2, '2024-06-02', 150.0),
    (3, '2024-06-01', 200.0)
]
sales_df = spark.createDataFrame(data, schema)

# Set the process_date variable to '2024-06-01'
# Your code here

Hint

Just assign the string '2024-06-01' to the variable process_date.

3

Filter data and remove existing output

Create a new DataFrame called filtered_df by filtering sales_df where the 'date' column equals process_date. Then, remove any existing parquet files in the folder output/2024-06-01 before writing.

Apache Spark

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

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

schema = StructType([
    StructField('id', IntegerType(), False),
    StructField('date', StringType(), False),
    StructField('amount', FloatType(), False)
])
data = [
    (1, '2024-06-01', 100.0),
    (2, '2024-06-02', 150.0),
    (3, '2024-06-01', 200.0)
]
sales_df = spark.createDataFrame(data, schema)

process_date = '2024-06-01'

# Filter sales_df for rows where 'date' equals process_date and assign to filtered_df
# Remove existing folder 'output/2024-06-01' if it exists
# Your code here

Hint

Use filter on sales_df with sales_df.date == process_date. Use shutil.rmtree to delete the folder if it exists.

4

Write filtered data and print count

Write filtered_df as parquet files to the folder output/2024-06-01. Then print the number of rows in filtered_df using print(filtered_df.count()).

Apache Spark

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

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

schema = StructType([
    StructField('id', IntegerType(), False),
    StructField('date', StringType(), False),
    StructField('amount', FloatType(), False)
])
data = [
    (1, '2024-06-01', 100.0),
    (2, '2024-06-02', 150.0),
    (3, '2024-06-01', 200.0)
]
sales_df = spark.createDataFrame(data, schema)

process_date = '2024-06-01'

filtered_df = sales_df.filter(sales_df.date == process_date)

output_path = f'output/{process_date}'
if os.path.exists(output_path):
    shutil.rmtree(output_path)

# Write filtered_df as parquet to output_path
# Print the count of rows in filtered_df
# Your code here

Hint

Use filtered_df.write.parquet(output_path) to save the data. Use print(filtered_df.count()) to show how many rows were written.

Practice

1. What does it mean for a data pipeline to be idempotent in Apache Spark?

easy

A. It requires manual cleanup after each run.

B. It runs faster each time it is executed.

C. It only processes new data and ignores old data.

D. It produces the same output even if run multiple times with the same input.

Solution

1. Step 1: Understand the meaning of idempotency in data pipelines

   Idempotency means running the pipeline multiple times with the same input should not change the output after the first run.

2. Step 2: Relate to Apache Spark pipeline behavior

   In Spark, an idempotent pipeline ensures consistent results by handling duplicates and overwriting data safely.

3. Final Answer:

   It produces the same output even if run multiple times with the same input. -> Option D

4. Quick Check:

   Idempotent = Same output every run [OK]

Hint: Idempotent means repeat runs give same results [OK]

Common Mistakes:

• Thinking idempotent means faster runs
• Assuming it only processes new data
• Believing manual cleanup is needed

2. Which Spark write mode helps achieve idempotency by replacing existing data safely?

easy

A. append

B. ignore

C. overwrite

D. update

Solution

1. Step 1: Review Spark write modes

   Append adds new data without removing old data, overwrite replaces existing data, ignore skips writing if data exists, update is not a standard Spark write mode.

2. Step 2: Identify which mode replaces data for idempotency

   Overwrite mode replaces the existing data, ensuring the output is consistent after multiple runs.

3. Final Answer:

   overwrite -> Option C

4. Quick Check:

   Overwrite mode = Replace data safely [OK]

Hint: Overwrite mode replaces data for consistent output [OK]

Common Mistakes:

• Choosing append which adds duplicates
• Selecting ignore which skips writes
• Thinking update is a valid write mode

3. Given this Spark code snippet, what will be the output count after running twice?

df = spark.createDataFrame([(1, 'a'), (2, 'b'), (1, 'a')], ['id', 'value'])
dedup_df = df.dropDuplicates()
dedup_df.write.mode('overwrite').save('/tmp/data')
loaded_df = spark.read.load('/tmp/data')
print(loaded_df.count())

medium

A. 2

B. 0

C. 1

D. 3

Solution

1. Step 1: Understand the input data and deduplication

   The input DataFrame has 3 rows with one duplicate (id=1, value='a'). dropDuplicates() removes duplicates, leaving 2 unique rows.

2. Step 2: Analyze the write and read operations

   Using overwrite mode saves the deduplicated data, replacing any previous data. Reading it back will load 2 rows.

3. Final Answer:

   2 -> Option A

4. Quick Check:

   Deduplication + overwrite = 2 rows [OK]

Hint: Deduplicate then overwrite keeps unique rows only [OK]

Common Mistakes:

• Counting duplicates as separate rows
• Assuming append mode adds duplicates
• Ignoring overwrite effect on saved data

4. You have a Spark pipeline that appends data without deduplication. What issue can this cause when rerunning the pipeline?

medium

A. Data duplication leading to incorrect results

B. Pipeline runs faster each time

C. Data is automatically cleaned up

D. Output files are overwritten safely

Solution

1. Step 1: Understand append mode without deduplication

   Appending data adds new rows without removing duplicates, so rerunning the pipeline adds repeated data.

2. Step 2: Identify the impact on results

   Repeated runs cause duplicated data, which can lead to wrong analytics or reports.

3. Final Answer:

   Data duplication leading to incorrect results -> Option A

4. Quick Check:

   Append without deduplication = duplicates [OK]

Hint: Appending without deduplication causes duplicates [OK]

Common Mistakes:

• Thinking pipeline runs faster
• Assuming data cleans automatically
• Believing output files overwrite safely

5. You want to design an idempotent Spark pipeline that processes daily sales data. Which combination ensures idempotency when rerunning for the same day?
1. Read raw daily data
2. Remove duplicates
3. Write results to storage
Choose the best write mode and deduplication approach.

hard

A. Use append mode and no deduplication

B. Use overwrite mode and remove duplicates before writing

C. Use ignore mode and append duplicates

D. Use append mode and remove duplicates after writing

Solution

1. Step 1: Identify deduplication timing

   Removing duplicates before writing ensures only unique data is saved, preventing repeated entries.

2. Step 2: Choose write mode for idempotency

   Overwrite mode replaces existing data for the day, so rerunning does not add duplicates.

3. Final Answer:

   Use overwrite mode and remove duplicates before writing -> Option B

4. Quick Check:

   Deduplicate + overwrite = idempotent pipeline [OK]

Hint: Deduplicate first, then overwrite to ensure idempotency [OK]

Common Mistakes:

• Appending duplicates causing repeated data
• Removing duplicates after writing is too late
• Using ignore mode skips writing data