Spark JDBC read is slow or runs in one task: partitionColumn, bounds, predicates and fetchsize

#Symptom

An ingestion job reads a large table from a relational database with spark.read.format("jdbc"). One of these happens:

• The Spark UI shows the read stage with one task running for an hour while every other core sits idle.
• One task out of many runs far longer than the rest.
• The executor fails with OutOfMemoryError, or the database DBA reports one huge full scan, or many sessions hammering the primary.

By default Spark opens one connection and runs one query for a JDBC read. Unless you tell it how to split the table, the whole table goes through a single task.

#Quick checks

1. In the Spark UI, open the stage for the JDBC scan and check the task count. One task means the read is not partitioned.
2. If there are many tasks, sort them by duration and input rows. A big gap between the median and the max means the split column is skewed.
3. Check the database side for active sessions from the Spark application user, their wait events and the SQL text.
4. Check the reader options in the code for partitionColumn, numPartitions and fetchsize.

df = spark.read.format("jdbc").options(**opts).load()
print(df.rdd.getNumPartitions())   # 1 = single connection, single query

#Fix 1: partitioned reads with a numeric or date column

opts = {
    "url": "jdbc:sqlserver://sql01.contoso.local:1433;databaseName=sales;encrypt=true",
    "dbtable": "dbo.orders",
    "user": dbutils.secrets.get("jdbc", "sales-user"),
    "password": dbutils.secrets.get("jdbc", "sales-pass"),
    "partitionColumn": "order_id",
    "lowerBound": "1",
    "upperBound": "250000000",
    "numPartitions": "16",
    "fetchsize": "10000",
}
df = spark.read.format("jdbc").options(**opts).load()

(dbutils.secrets is Databricks. Elsewhere, read the credentials from your own secret store.)

How it works:

• partitionColumn must be a numeric, date or timestamp column. partitionColumn, lowerBound, upperBound and numPartitions must be set together.
• Spark splits the range lowerBound–upperBound into numPartitions equal strides and runs one WHERE query per stride.
• The bounds do not filter rows. The first partition also takes everything below lowerBound and the last takes everything above upperBound. Wrong bounds therefore cause skew, not data loss.

Get real bounds from the source rather than guessing:

b = (spark.read.format("jdbc")
     .option("url", opts["url"]).option("user", opts["user"]).option("password", opts["password"])
     .option("query", "SELECT MIN(order_id) AS lo, MAX(order_id) AS hi FROM dbo.orders")
     .load().first())
opts.update(lowerBound=str(b.lo), upperBound=str(b.hi))

Warning

query cannot be combined with partitionColumn. To partition a custom query, pass it as a subquery in dbtable, for example "(SELECT ... FROM dbo.orders WHERE status <> 'X') AS src".

The split column must be indexed at the source. Without an index, each of the 16 partition queries does its own full scan, and the read gets slower than the single-task version.

#Fix 2: explicit predicates for skewed or non-numeric keys

When IDs are sparse (big gaps, or most rows in a recent range), equal strides produce badly unbalanced tasks. Pass your own list of WHERE clauses instead. There is one partition per predicate.

predicates = [
    f"order_date >= '{d0}' AND order_date < '{d1}'"
    for d0, d1 in [("2026-01-01", "2026-04-01"), ("2026-04-01", "2026-07-01"),
                   ("2026-07-01", "2026-09-01"), ("2026-09-01", "2026-10-01"),
                   ("2026-10-01", "2026-11-01")]
]
props = {"user": opts["user"], "password": opts["password"], "fetchsize": "10000"}
df = spark.read.jdbc(url=opts["url"], table="dbo.orders", predicates=predicates, properties=props)

Make the predicates non-overlapping and complete. Spark does not check them. Overlaps duplicate rows, and gaps silently drop them. Use half-open ranges (>= and <) and check the row count afterwards.

For tables without a usable numeric or date key, split on a hash of the key. Function names vary by engine:

Hash predicates cannot use an index, so each one scans the table. Use them on moderate tables or on a replica.

#Fix 3: fetch size

fetchsize controls how many rows the driver pulls per network round trip. Driver defaults are often small. The Oracle JDBC driver defaults to 10 rows. The PostgreSQL driver ignores the fetch size and loads the entire result set into memory unless autocommit is off, which Spark handles when fetchsize is set.

Raise it in steps and watch executor memory. A large fetch size with wide rows moves the bottleneck into an OutOfMemoryError.

#Fix 4: push work to the database and protect it

opts.update({
    "pushDownPredicate": "true",        # default true: .filter() becomes WHERE
    "queryTimeout": "3600",             # seconds per statement; 0 = no limit
    "sessionInitStatement": "SET TRANSACTION ISOLATION LEVEL READ COMMITTED",
})
df = (spark.read.format("jdbc").options(**opts).load()
      .filter("order_date >= '2026-10-01'")
      .select("order_id", "customer_id", "amount", "order_date"))
df.explain()   # check PushedFilters in the JDBC scan node

• Select only the columns you need. Spark pushes the column list into the SELECT.
• Check PushedFilters in explain(). Filters on expressions Spark cannot translate are applied after the full read.
• sessionInitStatement runs once per connection before the read. Use it for session settings such as isolation level or Oracle ALTER SESSION parameters. Pick an isolation level your DBA agrees with.
• customSchema can override the read type of awkward columns. For example, "customSchema": "amount DECIMAL(38,10)" fixes Oracle NUMBER columns without precision.

Caution

numPartitions is also the number of concurrent connections Spark opens against the source. Sixty-four parallel full-range queries on an OLTP primary can cause a production outage. Agree on a cap with the DBA, read from a replica or snapshot where possible, and schedule large extracts outside business hours.

#Validate the result

src_count = (spark.read.format("jdbc")
             .option("url", opts["url"]).option("user", opts["user"]).option("password", opts["password"])
             .option("query", "SELECT COUNT(*) AS n FROM dbo.orders WHERE order_date >= '2026-10-01'")
             .load().first().n)
print(src_count, df.count())

The counts should match for a static source. On a live OLTP table, each partition query runs in its own transaction at a slightly different time, so rows that change mid-read can be missed or duplicated. For exact loads, read from a database snapshot or replica, or bound the read with a high-water mark captured before the read starts.

#Prevent it next time

• Build a shared reader function that always sets fetchsize, a partitioning strategy and a connection cap. Don't let each job reinvent the options.
• Prefer incremental reads on an indexed modified_at or ID watermark over full-table pulls. Move to log-based CDC when the full pull no longer fits the window.
• Record the task count and the max-to-median task time for each JDBC read in job logs. Alert when a read collapses to one task.
• Keep the split column indexed. Re-check the bounds when the table grows beyond the old upperBound.
• Store credentials in a secret scope or key vault, never in the options dict in source control.