Spark SQL - Data Sources & Formats: Parallel Ingestion Workbook

This workbook walks you through configuring parallel JDBC ingestion queries step-by-step.

1. Cluster & DB Parameters

• Database Table: transactions (Primary Key: tx_id, range: 1 to 10,000,000)
• Ingestion Configuration:
  • partitionColumn: tx_id
  • lowerBound: 1
  • upperBound: 10,000,000
  • numPartitions: 5

2. Tasks

Task 1: Formulate the Partition SQL Queries

Based on the configuration above, calculate the exact range bounds for each of the 5 generated tasks. Write down the matching SELECT query constraints that Spark will execute on the database server.

Task 2: Handle Skewed Ingestion Keys

Assume the tx_id range is heavily skewed, with 90% of rows concentrated between 8,000,000 and 10,000,000. Detail why this skew degrades parallel JDBC ingestion, and propose a solution using database-side mod function mapping.

3. Step-by-Step Solutions

Solution 1: Range calculations

• Interval Size Formula:

Interval = (upperBound - lowerBound) / numPartitions
Interval = (10,000,000 - 1) / 5 = 2,000,000

• Query Tracing Matrix:
  • Task 1: Queries the first range block.
    • SQL: SELECT * FROM transactions WHERE tx_id < 2000001
  • Task 2:
    • SQL: SELECT * FROM transactions WHERE tx_id >= 2000001 AND tx_id < 4000001
  • Task 3:
    • SQL: SELECT * FROM transactions WHERE tx_id >= 4000001 AND tx_id < 6000001
  • Task 4:
    • SQL: SELECT * FROM transactions WHERE tx_id >= 6000001 AND tx_id < 8000001
  • Task 5: Handles the last range up to the upper boundary.
    • SQL: SELECT * FROM transactions WHERE tx_id >= 8000001
• All 5 tasks run concurrently, allowing executors to stream their partitions in parallel.

Solution 2: Handling Skewed Ingestion Keys

• The Skew Issue: If 90% of records reside in the final range 8,000,000 - 10,000,000, Task 5 will ingest 9 million rows, while Tasks 1-4 ingest only 250,000 rows each. The parallel import will stall, waiting for Task 5 to complete.
• The Modulo Solution: Replace the direct tx_id range split with a modulo hash function on the PK using a database-side inline SQL view or subquery:

# Partition using modulo calculations to distribute skewed records evenly
skew_sql = "(SELECT *, (tx_id % 5) AS part_key FROM transactions) AS tx_view"

df = spark.read.format("jdbc") \
    .option("url", "jdbc:sqlserver://host") \
    .option("dbtable", skew_sql) \
    .option("partitionColumn", "part_key") \
    .option("lowerBound", "0") \
    .option("upperBound", "4") \
    .option("numPartitions", "5") \
    .load()

This hashes every record into a balanced bucket (0 to 4), guaranteeing uniform partition loading.