Definition
- Default problem – Spark reads JDBC through one connection and one task, which is slow for large tables.
- Parallel reads – specify a numeric
partitionColumnwithlowerBound,upperBound, andnumPartitionsfor range-based parallel connections. - Fetchsize – increase rows per round trip (e.g., 10000) to cut network overhead.
- Pushdown – use a subquery as
dbtableto push filters and column selection to the database. - Writes – each task opens its own connection; no cross-task transaction, so partial writes are possible.
- Idempotency – write to a staging table and swap, or use upsert patterns.
The National Bank of India has a huge ledger in a vault. Sending one clerk through one door to copy every page is the default single-connection read. Optimizing is opening many doors and sending many clerks, each told to copy a range of pages. The fetchsize is how many pages each clerk carries per trip. For writes, each clerk has their own notebook with no single master transaction covering all of them.
Understanding the Concepts
Why is the default JDBC read slow?
By default, spark.read.jdbc() uses one connection and one task. A billion-row table streams through a single executor, and the database serves it from one session. This does not scale.
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("NBI-JDBC").getOrCreate()
# DEFAULT: single connection, single task (slow!)
df = spark.read \
.format("jdbc") \
.option("url", "jdbc:postgresql://nbi-db:5432/banking") \
.option("dbtable", "trades") \
.option("user", "etl_user") \
.option("password", "***") \
.load()
print(f"Partitions: {df.rdd.getNumPartitions()}") # 1!
Partitions: 1
How do you parallelize JDBC reads?
Specify four options together: a numeric partitionColumn, lowerBound, upperBound, and numPartitions. Spark divides the range into slices and issues one query per slice on its own connection.
# PARALLEL: 10 connections, each reading a range
df = spark.read \
.format("jdbc") \
.option("url", "jdbc:postgresql://nbi-db:5432/banking") \
.option("dbtable", "trades") \
.option("user", "etl_user") \
.option("password", "***") \
.option("partitionColumn", "trade_id") \
.option("lowerBound", "1") \
.option("upperBound", "10000000") \
.option("numPartitions", "10") \
.load()
print(f"Partitions: {df.rdd.getNumPartitions()}") # 10
# Task 1: WHERE trade_id >= 1 AND trade_id < 1000000
# Task 2: WHERE trade_id >= 1000000 AND trade_id < 2000000
# ...
Partitions: 10
How does fetchsize improve throughput?
The fetchsize controls how many rows the JDBC driver pulls per network round trip. The default is often tiny (10 rows), causing millions of round trips on large tables.
# Set fetchsize to 10000 rows per round trip
df = spark.read \
.format("jdbc") \
.option("url", "jdbc:postgresql://nbi-db:5432/banking") \
.option("dbtable", "trades") \
.option("partitionColumn", "trade_id") \
.option("lowerBound", "1") \
.option("upperBound", "10000000") \
.option("numPartitions", "10") \
.option("fetchsize", "10000") \
.load()
# Default fetchsize=10: 1M rows = 100K round trips per task
# fetchsize=10000: 1M rows = 100 round trips per task (1000x fewer!)
How do you push filters to the database?
Use a subquery as dbtable to push column selection and filters to the database so it sends only the data you need.
# Push filter and column selection to the database
df = spark.read \
.format("jdbc") \
.option("url", "jdbc:postgresql://nbi-db:5432/banking") \
.option("dbtable",
"(SELECT trade_id, amount, trade_date "
"FROM trades "
"WHERE status = 'SETTLED' "
"AND trade_date >= '2024-01-01') t") \
.option("partitionColumn", "trade_id") \
.option("lowerBound", "1") \
.option("upperBound", "10000000") \
.option("numPartitions", "10") \
.option("fetchsize", "10000") \
.load()
# The database filters and selects columns BEFORE sending data
# Much less data crosses the network
How do JDBC writes work in Spark?
Spark partitions the DataFrame and each task opens its own JDBC connection, inserting rows in batches. There is no single transaction across all tasks.
from pyspark.sql import functions as F
result = df.groupBy("trade_date").agg(F.sum("amount").alias("total"))
# Write to the database
result.write \
.format("jdbc") \
.option("url", "jdbc:postgresql://nbi-db:5432/banking") \
.option("dbtable", "daily_totals") \
.option("user", "etl_user") \
.option("password", "***") \
.option("batchsize", "10000") \
.mode("overwrite") \
.save()
# Each task opens its own connection and commits independently.
# If the job fails after 5 of 10 tasks, 5 partitions are committed.
# The table is left PARTIALLY written!
Why do writes need idempotency?
Because there is no cross-task transaction, a failed job can leave the target partially written. Design for idempotency.
# WRONG: write directly to the production table
result.write.mode("overwrite").jdbc(url, "daily_totals", props)
# If job fails mid-write, table is half-populated.
# RIGHT: write to staging, then swap atomically
result.write.mode("overwrite").jdbc(url, "daily_totals_staging", props)
# Then in the database:
# BEGIN;
# DROP TABLE IF EXISTS daily_totals_old;
# ALTER TABLE daily_totals RENAME TO daily_totals_old;
# ALTER TABLE daily_totals_staging RENAME TO daily_totals;
# COMMIT;
How do you control write connections?
Each partition opens a connection. Too many partitions can exhaust the database connection pool.
# WRONG: 200 partitions = 200 simultaneous connections!
result.write.jdbc(url, "daily_totals", props)
# RIGHT: coalesce before writing to limit connections
result.coalesce(10).write.jdbc(url, "daily_totals", props)
# Only 10 simultaneous connections
What about partition column choice?
The partition column should have an even distribution. A skewed column causes one task to read most of the data while others are idle.
# GOOD: auto-increment trade_id has even distribution
.option("partitionColumn", "trade_id")
# BAD: branch_id has 90% of data in one branch
.option("partitionColumn", "branch_id")
# One task reads 90% of the data!
# Note: lowerBound and upperBound do NOT filter data.
# Values outside the range land in the first or last partition.
What should a senior engineer know?
A senior engineer reads from a read replica to avoid loading the production primary, stages writes for atomicity, and monitors connection pool usage.
# Senior JDBC checklist:
# 1. Parallel reads: partitionColumn + bounds + numPartitions
# 2. fetchsize: 5000-10000 (tune per database)
# 3. Pushdown: use subquery as dbtable
# 4. Read from replica, not primary
# 5. Coalesce before writes to limit connections
# 6. Stage writes + swap for atomicity
# 7. Monitor database connection pool usage
Visual Overview
Default JDBC read (single connection):
Spark ─── 1 connection ───→ Database
(1 task, 1M rows, slow) (1 session)
Parallel JDBC read (10 connections):
Spark ─── conn 1 ───→ WHERE id >= 0 AND id < 1M
─── conn 2 ───→ WHERE id >= 1M AND id < 2M
─── conn 3 ───→ WHERE id >= 2M AND id < 3M
... ...
─── conn 10 ──→ WHERE id >= 9M AND id < 10M
(10 tasks in parallel, 10x faster)
JDBC write (no cross-task transaction):
Task 1 ──→ INSERT batch 1 → COMMIT ✓
Task 2 ──→ INSERT batch 2 → COMMIT ✓
Task 3 ──→ INSERT batch 3 → FAIL ✗
Task 4 ──→ (never started)
Result: partial write! Tasks 1-2 committed, 3-4 not.
Key Takeaway
By default, Spark reads JDBC through one connection and one task, which does not scale. The key optimization is parallel reads: specify a numeric partition column with lower and upper bounds and a partition count, and Spark divides the range into slices, each on its own connection. Alongside that, increase fetchsize to thousands of rows per round trip, and push filters into the SQL by using a subquery as dbtable.
On the write side, each task opens its own connection and commits independently. There is no cross-task transaction, so a failed job leaves the table partially written. Design writes for idempotency by writing to a staging table and swapping, or using upsert patterns. Coalesce before writing to avoid opening more connections than the database pool can handle.
At the National Bank of India, our nightly analytics job read a hundred-million-row trades table. The single-connection default took hours. Setting a trade_id partition column with 10 partitions and a fetchsize of 10000, pointed at a read replica, brought the read to minutes. On the write side, we wrote to a staging table and swapped atomically after a mid-job failure once left a half-written report.
Remember one sentence: parallelize JDBC reads with a partition column and bounds, tune fetchsize for throughput, push filters to the database, and design writes for idempotency because each task is its own transaction.
---