Content

1. Background
1. 1.1 Overview
2. 1.2 The problems
2. Problem analysis
3. Technical solution
4. Cost & performance notes
5. Test focus
6. Other issues

1. Background

1.1 Overview

We ingest event data from a source system into the bronze layer of a medallion architecture on Databricks. Each batch arrives as JSON (JSON Lines) files in a raw storage container, and every file contains messages for many different topics. A single notebook reads the whole batch, splits it per topic, flattens the nested payloads, and writes each topic (and its nested child tables) into a Databricks-managed Delta table in the bronze schema.

Two batch shapes dominate:

First / historical load: 8,533 JSON files in one batch.
Daily delta load: ~200 JSON files per batch.

The whole pipeline runs on serverless compute for jobs, which changes a couple of assumptions you might carry over from classic clusters.

The historical load of 8,533 files is the worst case and the one we tuned against. Starting from a naive sequential implementation, two changes took the end-to-end runtime from ~5 hours to ~18 minutes:

Version	Runtime (8,533 files)
Sequential baseline	~5 h
Parallel job (thread pool + per-table lock)	~2 h
Parallel job + materialized temp table	~18 m

1.2 The problems

Reading 8,533 JSON files is expensive, and the DataFrame is read many times — once per topic, plus counts and schema inference. On a classic cluster you would just df.cache(). On serverless that lever is effectively gone (see section 3.2), so every downstream action re-reads and re-parses all 8.5k files.
Topics are processed in a loop. Each topic does latency-bound work: per-table Delta commits and validation checks. Doing this strictly sequentially leaves the cluster idle while the driver waits on commits, so wall-clock time (and therefore cost) is higher than it needs to be.
When we parallelize topic processing, different topics can produce the same nested child table (e.g. currentprice). Concurrent Delta appends to the same table raise DELTA_CONCURRENT_APPEND.

2. Problem analysis

The pipeline has two independent bottlenecks, and each needs its own fix:

Repeated reads of the source files. The root DataFrame df_raw_load is consumed once per topic. With 8,533 files that is thousands of files re-scanned and re-parsed on every single action. We need to read the files once and reuse the result cheaply.
Sequential, latency-bound writes. Each topic writes several Delta tables and runs validation. These operations spend most of their time waiting (commit, metadata, checks), not computing. That is exactly the profile where concurrency helps — while topic A commits, topic B can compute.
Shared child tables under concurrency. Parallelism introduces a correctness hazard: two threads appending to the same Delta table at the same time. This must be serialized per table, not globally (global serialization would throw away the benefit).

3. Technical solution

3.1 Parallel topic handling + per-table lock

We process topics concurrently with a ThreadPoolExecutor. Each task filters the source DataFrame down to one topic, flattens it, and writes it out. To avoid DELTA_CONCURRENT_APPEND, we keep one lock per target table name: parallel topics that happen to write the same table are serialized, while writes to different tables run fully in parallel.

Key details:

The lock is created lazily per table via dict.setdefault, guarded by a single mutex so the dictionary itself is thread-safe.
The DataFrame is passed as an argument to the worker (not read from a global), so each task is self-contained.
Run-control rows are collected in memory during the parallel phase and written once, sequentially, afterwards — that keeps the run-control Delta table free of concurrent writes too.

import threading
from concurrent.futures import ThreadPoolExecutor, as_completed

# one lock per target table: parallel topics can produce the SAME nested child table
# (e.g. currentprice), and concurrent Delta appends to one table raise
# DELTA_CONCURRENT_APPEND -> serialize writes per table
_table_write_locks = {}
_table_write_locks_guard = threading.Lock()

def _get_table_write_lock(table_name):
    with _table_write_locks_guard:
        return _table_write_locks.setdefault(table_name, threading.Lock())

def write_dataframe_to_table(df, table_name):
    df_prepared = table_writer_client.prepare_dataframe(df=df, table_name=table_name, ...)
    validation_client.execute_pipeline_checks(df=df_prepared, table_name=table_name, ...)

    # only writes to the SAME table are serialized; different tables run in parallel
    with _get_table_write_lock(table_name):
        table_writer_client.write_dataframe(df=df_prepared, table_name=table_name, ...)

Driving it with a bounded pool:

MAX_PARALLEL = 32  # tune to workload; cap to the number of topics

run_control_records = []
with ThreadPoolExecutor(max_workers=MAX_PARALLEL) as pool:
    futures = {pool.submit(handle_topic, t, df_raw_load): t for t in topic_names}
    for fut in as_completed(futures):
        run_control_records.extend(fut.result())

# write run_control rows ONCE, sequentially, after the parallel section
for rec in run_control_records:
    run_control_agent.insert_row(build_run_control_row(rec))

Why per-table and not per-run? A global lock would serialize everything and remove the speedup. A per-table lock only pays the cost when there is a genuine conflict (two topics writing the same child table), which is the only case that actually triggers DELTA_CONCURRENT_APPEND.

Tuning MAX_PARALLEL. It is a cap, not a target. If you have fewer topics than the cap, effective parallelism equals the topic count. Setting it too high on the heavy 8k load can pressure the driver (each thread does .collect() / .count()) and push serverless to autoscale harder, so benchmark before raising it. A safe default is min(32, len(topic_names)).

3.2 No cache on serverless — materialize into a temp Delta table

On classic clusters you would keep the parsed rows hot with df.cache() / persist(). On serverless compute you cannot rely on RDD caching, so a cached plan does not save you from re-reading the 8,000 JSON files on every action. Every .count(), every per-topic .filter(...), every schema inference re-scans and re-parses the raw files.

The fix is to read the files once and write them to a temporary managed Delta table, then read that table for all downstream work. Delta is columnar, already-parsed, and cheap to filter — so the expensive 8k-file scan happens exactly once.

# read + parse the 8k JSON files ONCE
df_raw_load = (
    spark.read.option("multiline", "false")
    .schema(main_schema).json(raw_file_list)
    .select(*renamed_metadata_cols, F.col("Payload"))
    .dropDuplicates(dedup_cols)
)

# serverless has no RDD cache; materialize once into a temp delta table so the
# json files are read only once and reused across all topic reads
temp_table_name = f"_tmp_raw_load_{batch_id}"
# backtick-quote each identifier part because the catalog name can contain hyphens
temp_table_fqn = f"`{catalog_name}`.`{bronze_schema}`.`{temp_table_name}`"

df_raw_load.write.format("delta").mode("overwrite") \
    .option("overwriteSchema", "true").saveAsTable(temp_table_fqn)

# everything downstream reads the cheap, columnar Delta table instead of 8k JSON files
df_raw_load = spark.read.table(temp_table_fqn)

And clean it up once the batch is done:

# drop the temp materialization table now that all topics have been processed
spark.sql(f"DROP TABLE IF EXISTS {temp_table_fqn}")

Two implementation notes that bit us:

Backtick every identifier part. Our catalog name contains hyphens (we-pensioendp-o-uc). Without backticks, saveAsTable throws INVALID_IDENTIFIER. Quote each of catalog / schema / table separately.
Scope the temp table name per batch_id. Concurrent batch runs then never collide on the same temp table, and cleanup is unambiguous.

3.3 How the two fixes combine

The materialized temp table (3.2) is also what makes the parallelism (3.1) cheap and safe: every worker thread runs df_raw_load.filter(topic == ...) against the Delta table, not against the raw JSON. So the 8,533-file scan is paid once, up front, and the 32 concurrent workers only do cheap columnar filters plus their writes.

flowchart TD
    A[8,533 JSON files in raw storage] -->|read + parse ONCE| B[Temp Delta table  _tmp_raw_load_batchid]
    B --> C{ThreadPoolExecutor  max_workers=32}
    C --> D[handle_topic A]
    C --> E[handle_topic B]
    C --> F[handle_topic ...]
    D --> G[per-table lock -> Delta append]
    E --> G
    F --> G
    G --> H[Bronze managed tables]
    H --> I[Drop temp table]

4. Cost & performance notes

On the 8,533-file historical load the two fixes compounded:

Version	Runtime	Delta vs previous
Sequential baseline	~5 h	—
Parallel job (thread pool + per-table lock)	~2 h	~2.5x faster
Parallel job + materialized temp table	~18 m	~6.7x faster

Overall that is roughly a 16x speedup (5 h → 18 m). Serverless is billed by DBU consumption over time (roughly compute allocated x wall-clock time), with autoscaling and Photon on by default. The total work (parse + flatten + write) is fixed; both fixes reduce wall-clock time rather than the work itself:

Parallelism (5 h → 2 h): overlaps latency-bound Delta commits/validation across topics, cutting idle wall-clock. The per-table lock keeps it correct without serializing everything.
Materialized temp table (2 h → 18 m): turns thousands of repeated 8,533-file scans into one, then cheap columnar Delta reads. This was the dominant win, because on serverless there is no RDD cache to fall back on.
Daily 200-file load: parallelism still helps (per-topic latency dominates), but absolute cost is small either way.

Rule of thumb: keep both, cap MAX_PARALLEL to the topic count, and confirm with the billing system table (system.billing.usage) by comparing a run at MAX_PARALLEL=32 vs a lower value.

5. Test focus

Correctness under concurrency: force two topics that emit the same nested child table (e.g. currentprice) and confirm no DELTA_CONCURRENT_APPEND, and that row counts match the sequential baseline.
Single read: verify the raw JSON path is scanned once (check the query profile / input size) and that per-topic filters hit the temp Delta table.
Temp table lifecycle: confirm the temp table is created per batch_id and dropped at the end, even across overlapping batch runs.
Identifier quoting: run against a catalog whose name contains hyphens and confirm saveAsTable succeeds.
Cost regression: capture DBUs for the 8,533-file load at a few MAX_PARALLEL values and pick the knee of the curve.

6. Other issues

Should the temp-table drop be best-effort (log a warning) or hard-fail if it cannot delete? Leaning best-effort so a cleanup hiccup does not fail an otherwise successful load.
Auto-tune MAX_PARALLEL from len(topic_names) and batch size instead of a fixed 32.
Consider partitioning / Z-ordering the temp table by topic_name so per-topic filters prune even more aggressively on the 8,533-file load.

Loading thousands of JSON files into a Databricks managed table on serverless

Categories

Tags

Table of Contents