Tuning Spark Jobs From the Transaction Log

When Databricks compute is slow, the quick fix is to enable Photon on it. Photon is Databricks' vectorized C++ execution engine, and for the right workloads it's a genuine multiplier. Photon accelerates execution by processing rows in vectorized batches instead of one at a time. Its Predictive I/O feature also reaches further than execution alone, using learned access patterns to skip unnecessary reads. Adaptive Query Execution (AQE) fixes the plan mid-flight, reshuffling joins and partitions once it sees real data.

None of this, though, touches the layer underneath: which files exist, what stats they carry, and how the table's physical layout got that way. That's governed entirely by the Delta transaction log. Predictive I/O can skip reading parts of a file more intelligently once that file is already in play, but it can't decide the file never needed to exist, can't invent a statistic the log never recorded, and can't undo a small-file problem baked into the table's layout. So there's a whole class of slowness that neither Photon nor AQE can touch, and the log is where you both diagnose it and fix it. This post walks through those controls, one at a time.

What's in _delta_log/ — and How We'll Read It

Every Delta table is just a folder of Parquet data files with one special sub-folder next to them: _delta_log/. That folder is the table's brain — it's the authoritative record of which files currently belong to the table, what's in them, and everything that ever happened.

my_table/
├── part-00000-....snappy.parquet   ← actual data
├── part-00001-....snappy.parquet
└── _delta_log/
    ├── ...0000.json                ← commit 0, one action/line
    ├── ...0001.json                ← commit 1
    ├── ...
    ├── ...0010.checkpoint.parquet  ← snapshot every 10 commits
    └── _last_checkpoint            ← newest-checkpoint pointer

The below are the two fundamental things to remember when it comes to the transaction log:

• Each numbered .json file is exactly one commit, and each line inside it is one action, a small JSON object describing one thing that happened.
• The data files are never modified in place. A change means writing new files and marking the old ones as removed in the log. The log itself is append-only too: each commit is a new JSON file, numbered in sequence, so together they hold a complete history of the physical layout, not just its current state.

The actions that matter for performance tuning:

import json, glob, os

def read_commit(log_dir, version):
    with open(f"{log_dir}/{version:020d}.json") as f:
        # one action per line
        return [json.loads(line) for line in f]

Note: Unity Catalog managed tables block direct workspace access to raw files, so these techniques assume an external table with storage you own. On a managed table, fall back to dt.history(), dt.detail(), and _metadata — enough to apply fixes, less to see why. You can still read the files directly if your Identity and Access Management (IAM) principal has provider-level access to the underlying storage.

How Photon Actually Works — and Where It Stops

Photon replaces Spark's default Java Virtual Machine (JVM) execution engine. It's a native C++, columnar, vectorized execution engine:

• It processes data in batches of columns, not one row at a time — so a filter or aggregate runs down a tight array of values.
• That layout lets the CPU process many values at once instead of one at a time, and keeps related data close together so it's faster to access.
• Native memory management means no JVM garbage-collection pauses mid-scan.

For heavy CPU work like scanning, filtering, joining, and aggregating rows you actually need, Photon is a real speedup, sometimes several times over.

Now look at what Photon actually gets handed. By the time it runs, the planner has already decided which files to open and which rows survive pruning. Photon starts after those decisions are made. It makes the rows it's given cheap to chew through, but it has no say in how many rows it's given. It can't skip a file. It can't compact small files. It can't create a statistic that was never collected. It can't undo a full file rewrite from an earlier update.

So when the real bottleneck is how much gets read, how many files exist, missing statistics, or slow planning, Photon just burns through work fast that never needed to exist in the first place. That work gets decided one layer down, in the log.

The Layer Cake: Photon, the Spark Optimizers, and the Log

"Just let Photon (or AQE, or the cost-based optimizer) handle it" is the reflex — so it's worth seeing exactly where each one lives. A query's wall-clock time is a stack, and Photon and every Spark optimizer sit in the upper layers, all fed by a bottom layer only the log controls:

• Photon executes the plan it's given.
• AQE reshapes shuffles at runtime — it coalesces small shuffle partitions and fixes skew. AQE fixing small shuffle partitions is not the same as OPTIMIZE fixing small data files.
• Catalyst pushes your WHERE down to the Delta scan — but once pushed, the filter is answered by the log's min/max stats. Catalyst does the handoff; the log does the skipping.

So "tuning" splits cleanly:

• Photon and the Spark optimizers make the given work faster.
• The log changes what work exists — and every layer above inherits that change for free.

How the Delta Log Comes Into Play

1. Pruning and Clustering: Make Ranges Narrow Enough to Skip Files

A file only gets skipped if its stats rule it out for your filter. Two things can break that: the column has no stats at all, or the column has stats but every file's range spans the whole domain, so nothing narrows anything.

Every add action carries a per-file stats blob. Take agency, a column in this table, as the example:

adds = (a["add"] for a in read_commit(LOG_DIR, 5) if "add" in a)
s = json.loads(next(adds)["stats"])

print("indexed?", "agency" in s["minValues"])
print(s["minValues"].get("agency"), "to",
      s["maxValues"].get("agency"))

If the column is missing from minValues, it was never indexed. Delta only indexes the first 32 columns by default (delta.dataSkippingNumIndexedCols), so a filter on column 48 in a wide bronze table scans everything no matter how selective it looks. If the column is present but the range is wall-to-wall (A to Z in every file), the stats exist but every file overlaps your filter anyway. That's not a missing-stats problem, it's a layout problem: similar values are scattered everywhere instead of grouped together.

What to do:

• Filter on a column you already partition or cluster by, before reaching for anything else.
• Index a late column explicitly: ALTER TABLE ... SET TBLPROPERTIES ('delta.dataSkippingStatsColumns' = '...') (only affects files written after the change).
• Fix the layout itself with OPTIMIZE ... ZORDER BY (agency) or ALTER TABLE ... CLUSTER BY (agency) for Liquid Clustering, so ranges become narrow and disjoint instead of overlapping.

How to confirm it: Re-read add.stats on freshly written files. The column should now appear in minValues/maxValues, and ranges should tighten from overlapping ("A" → "Z") to disjoint ("A" → "M", "N" → "Z"). Files pruned should visibly jump in the query profile.

2. Small Files: Every File Carries a Fixed Cost to Open

Each file is a storage open and a task to schedule, regardless of how few rows it holds. Frequent small writes, especially MERGE, tend to produce a lot of tiny files over time.

File count isn't visible in any single snapshot. It only shows up as a trend across commits.

for path in sorted(glob.glob(f"{LOG_DIR}/*.json")):
    v = int(os.path.basename(path).split(".")[0])
    acts = [json.loads(l) for l in open(path)]
    adds = [a["add"] for a in acts if "add" in a]
    rems = sum("remove" in a for a in acts)
    info = (a["commitInfo"] for a in acts if "commitInfo" in a)
    op = next(info, {}).get("operation", "?")
    tiny = sum(1 for a in adds if a["size"] < 1_000_000)
    print(f"v{v:<3} {op:<10} "
          f"+{len(adds)}/-{rems} ({tiny} under 1MB)")

v5  MERGE      +20/-16 (20 under 1MB)
v6  MERGE      +19/-2  (19 under 1MB)
v10 OPTIMIZE   +3/-78  (0 under 1MB)

Two things jump out. First, the file count keeps climbing across the MERGE commits, and almost every file added is under 1MB. Second, OPTIMIZE at v10 is the reverse: it removed 78 tiny files and left just 3 behind. That climb-then-collapse pattern is the fragmentation signature.

What to do:

• Run OPTIMIZE as a deliberate compaction pass.
• Set delta.autoOptimize.optimizeWrite = true to bin-pack at write time, and autoCompact = true to coalesce after writes.
• Turn on delta.enablePredictiveOptimization = true and let Databricks schedule this automatically.
• If the root cause is over-partitioning a low-cardinality column, stop partitioning it. That's a design fix the file counts point to.

How to confirm it: Check dt.detail().numFiles before and after, or watch for a large negative remove count in the OPTIMIZE commit.

3. MERGE Write Amplification: A Few Changed Rows, a Lot of Rewritten Data

Under copy-on-write, changing one row rewrites the entire file it lives in. A MERGE that touches a small number of rows can still rewrite a large volume of data if those rows are spread across many files.

commits = read_commit(LOG_DIR, 5)
info = (a["commitInfo"] for a in commits if "commitInfo" in a)
m = next(info)["operationMetrics"]

print(m["numTargetRowsUpdated"], "rows changed")
print(m["numTargetFilesRemoved"], "to",
      m["numTargetFilesAdded"], "files rewritten")

70 rows changed
16 to 20 files rewritten

70 rows changed, 16 files rewritten. If we assume 64MB a file, that's roughly 1GB moved to change 70 rows. Cross-checking those 16 files' key ranges against the incoming batch (the same stats from lever 1) usually shows the ranges overlap broadly, which is why Delta couldn't localize the change.

What to do:

• Add a predicate to the ON clause that maps to a partition or clustering column, so old partitions get skipped entirely before any rewrite happens.
• Cluster or Z-order the target on the merge keys so matching rows sit together in fewer files.
• Turn on delta.enableDeletionVectors = true so a small update marks rows instead of rewriting whole files.

How to confirm it: Re-run and check operationMetrics again. numTargetFilesRemoved should drop sharply, and numTargetDeletionVectorsAdded should rise above zero if deletion vectors are on.

4. Query Startup: How Far Back a Reader Has to Replay

Before any query runs, a reader has to reconstruct the table's current state by replaying the log from the last checkpoint forward. If that checkpoint is old and hundreds of commits have piled up since, the reader pays for all that replay before the query even starts.

last = json.load(open(f"{LOG_DIR}/_last_checkpoint"))
print("checkpoint at v", last["version"])

A large gap between the checkpoint version and the newest commit means every startup walks a lot of JSON. A bloated active-file count (dt.detail().numFiles) also means a heavier checkpoint, which loops back to the small-files problem above.

What to do:

• Keep the active-file count down. Fewer files means a smaller, faster checkpoint.
• delta.checkpoint.writeStatsAsStruct = true keeps pruning stats in a fast, columnar form inside the checkpoint. Checkpoints themselves are written automatically every 10 commits by default.

How to confirm it: _last_checkpoint version should sit close to the table's current version.

5. Skew: An Uneven Distribution Baked Into the Files

One file or partition holding far more rows than the rest makes a single task run long after everything else finishes.

sizes = [
    (json.loads(a["add"]["stats"])["numRecords"],
     a["add"]["path"][:30])
    for a in read_commit(LOG_DIR, 5) if "add" in a
]
for n, path in sorted(sizes, reverse=True)[:5]:
    print(n, path)

812004 part-00000-...
3      part-00002-...

An uneven spread like that, or one partition value dominating the counts, is skew, quantified before you rerun anything.

What to do:

• Pick a higher-cardinality or more even partition or clustering key.
• Repartition or salt the skewed key before writing, so rows spread across files more evenly.

How to confirm it: Re-read numRecords across files after the rewrite. The spread should flatten out.

The One Honest Caveat (Unity Catalog Managed Tables)

Everything that reads the raw log files directly — add.stats, commitInfo metrics, _autostats, _last_checkpoint — needs path-based access to _delta_log/. On a Unity Catalog managed table, that specific access pattern is blocked: the managed storage location lives under a reserved path (__unitystorage/catalogs/<id>/tables/<id>/), and any direct list or read against it fails with INVALID_PARAMETER_VALUE.LOCATION_OVERLAP, because catalog and schema storage locations are reserved for managed storage and don't support path-based access (Databricks KB; Databricks docs).

The log itself is still there. What you lose on a managed table is direct access to its raw JSON, not the log's existence. In its place you get the API and SQL layer: dt.history(), dt.detail() (numFiles, sizeInBytes, properties), the _metadata file-layout trick, the Spark UI query profile, and every ALTER TABLE / OPTIMIZE / ANALYZE fix covered above, all of which read the log internally without exposing it to you directly.

Either way the point stands: Photon is a faster engine for the work you do. The transaction log is how you stop doing work you never needed to. The biggest speedups usually aren't a hotter engine — they're the files Photon never had to open, decided one layer up, in a log that's been recording the whole time.

Next time a job is slow, before reaching for a bigger cluster or toggling Photon: open the log. Count the files. Read the stats. Half the time the answer — too many files, no stats on the filter column, an ON clause that rewrites the world — has been sitting in a .json file the entire time.