Databricks TPC-DI

A Databricks-native implementation of the TPC-DI data-integration benchmark, designed to run end-to-end on the Databricks Lakehouse platform across multiple compute SKUs (job clusters, SQL warehouses, Spark Declarative Pipelines, dbt-databricks) and multiple data-load shapes (single-batch, incremental, and a 365-day daily-streaming "Augmented Incremental" variant).

This repo follows the TPC-DI v1.1.0 spec for business rules and table outputs; the spec itself does not provide code, only requirements. This project is the implementation.

Notable changes vs the original DIGen-only pipeline

• Distributed Spark data generator (src/tools/data_gen_tasks/ entry + src/tools/tpcdi_gen/ modules) — replaces the single-threaded DIGen.jar with a parallel PySpark implementation that runs on Databricks Serverless. Linear scaling across executors; large scale factors (SF=10000+) finish in a fraction of the JAR's time. Decomposed into per-dataset workflow tasks (gen_* + copy_*) so a failed dataset can be repair-run in isolation. Trade family is further split into gen_trade_base + 4 parallel leaves (gen_trade, gen_tradehistory, gen_cashtransaction, gen_holdinghistory) sharing a Delta-staged base DataFrame. CustomerMgmt scheduling runs as a Pandas-UDF GrowingOffsetPermutation on Spark executors (no driver-side numpy bottleneck). File rename from staging→final layout fans out across executor pods. The DIGen.jar path is preserved for byte-compatible reference output. SF=20k full pipeline: ~19m on serverless.
• SDP (Spark Declarative Pipelines) — the runtime previously branded "DLT". Library names, schema labels, and prose all reflect the rename.
• Augmented Incremental benchmark — a 365-day daily-streaming reshaping of TPC-DI (2016-07-06 → 2017-07-05) that exercises CDC + SCD2 maintenance under a real production-shaped daily load instead of a single bulk import. Available for Cluster (job-cluster MERGEs), SDP (Spark Declarative Pipelines), and dbt (dbt-databricks against a DBSQL warehouse).
• Python workflow builders — every job/pipeline JSON is built by a Python module under src/tools/workflow_builders/. Jinja templates retired.
• Static audit snapshots — pre-computed *_audit.csv snapshots committed to the repo at every common SF, so audit values are instant rather than recomputed each run.
• Skills-asset positioning — this repo is deliberately curated for use by AI agents (Claude, Databricks Genie). See CLAUDE.md at repo root for architecture context, gotchas, and load-bearing decisions.

Flow at a glance

Choosing a workflow — SKU × Batch Type

The Driver splits the workflow choice into two widgets so the dropdown stays short. Pick a SKU (compute shape) and a Batch Type (how data is fed in). Combinations the SKU doesn't support are hidden automatically.

SKU \ Batch Type	Single Batch	Incremental	Augmented Incremental
Cluster (job cluster, classic or serverless)	✓	✓	✓
DBSQL (serverless SQL warehouse)	✓	✓	—
SDP (Spark Declarative Pipelines)	✓ + edition	—	✓
dbt (dbt-databricks against a SQL warehouse)	—	—	✓

When SKU=SDP × Batch Type=Single Batch, an Edition dropdown appears: CORE, PRO (adds APPLY CHANGES INTO for SCD Type 1/2), or ADVANCED (adds Data Quality constraints).

Batch Type detail

• Single Batch — all 3 TPC-DI batches in one pass (faster, no audit checks).
• Incremental — batches sequentially with audit checks at each boundary (spec validation, CLUSTER + DBSQL only).
• Augmented Incremental — 365-day daily streaming pipeline (Cluster, SDP, or dbt). Reads pre-staged per-day files from _staging/sf={sf}/{Dataset}/_pdate={date}/. Stage 0 (data prep) is a separate workflow built by workflow_builders/augmented_staging.py — run it once per SF to populate the staging tree. Validated at SF=10/100/1000/5000/10000/20000. Layout is uniformly Liquid clustering; the setup notebook for each variant DEEP CLONEs the per-SF staging schema and inherits its CLUSTER BY directly — see src/incremental_batches/augmented_incremental/README.md for the per-variant setup matrix and architectural overview. Per-variant deep-dives live next to their code: Cluster (job clusters), SDP, dbt.

How to run

1. Open src/TPC-DI Driver in your Databricks workspace.
2. Run the first cell to bootstrap defaults (cloud detection, node-type catalog, DBR list, user-prefixed schema names).
3. Set the widgets:
   • SKU + Batch Type + (if SDP × Single Batch) Edition
   • Scale Factor (10 / 100 / 1000 / 5000 / 10000 / 20000)
   • Job Name, Target Catalog, Target Database — reasonable defaults.
   • Data Generator (spark default; digen for the legacy single-threaded path) — hidden for Augmented variants since they always use Spark-staged data.
   • Serverless (YES default) — on for everything except Cluster non-serverless and SDP non-serverless.
   • Predictive Optimization, Optimize For UC Features or Fastest Performance.
4. Run the next cells. The Driver creates:
   • One datagen job (skipped for Augmented variants — they share a separate augmented_staging Stage 0 workflow).
   • One benchmark job for standard variants (Cluster / DBSQL / SDP single-batch or incremental).
   • Or a parent + child pair for Augmented Cluster; parent + child + pipeline for Augmented SDP; parent + child + DBSQL warehouse for Augmented dbt.

Each cell prints a clickable link to the created job(s).

Job naming convention

{base}-SF{sf}-{Batched}-{Exec}-{Gen} for standard variants; {base}-SF{sf}-AugmentedIncremental-{Cluster|SDP|DBT}-Parent for the augmented parent. Each job carries a data_generator: spark|native_jar tag so they're filterable without parsing the name.

Compute & sizing

Serverless (default)

SDP pipelines, augmented variants, and the Spark datagen all run on serverless with performance_target: PERFORMANCE_OPTIMIZED. No DBR / node-type / cluster restrictions.

Non-serverless Cluster (when Serverless=NO)

The Driver picks an ARM-preferred, local-NVMe-preferred node and sizes the cluster by SF. Cloud-aware (Azure D_v6 / L_v3, AWS Graviton m8g/m8gd/i7i, GCP c4a-lssd):

Scale Factor	Total Raw Data	Suggested Cluster
10	~1 GB	single-node 8-core
100	~10 GB	single-node 8-core
1000	~100 GB	single-node 16-core
5000	~500 GB	32-core driver + 5 × 16-core workers
10000	~1 TB	64-core driver + 10 × 16-core workers
20000	~2 TB	64-core driver + 20 × 16-core workers

DBSQL Warehouses

Auto-created if missing (serverless, sized by SF). Names are generic so multiple users can share. Two sizing tables apply:

Single-batch DBSQL workflow (heavy one-shot rewrite per batch):

SF	Warehouse
10 / 100	TPCDI_2X-Small
1000	TPCDI_Small
5000	TPCDI_Large
10000	TPCDI_X-Large

Augmented Incremental dbt (per-day MERGE/INSERT — much lighter):

SF range	Warehouse size
≤ 5,000	2X-Small
5,001 – 10,000	X-Small
10,001 – 20,000	Small (anchor — what we tuned for)
20,001 – 40,000	Medium
40,001 – 80,000	Large
…doubling…	…one size up…

Native DIGen.jar

Forced to a non-serverless DBR 15.4 + Photon cluster (Java subprocess can't run on serverless). The Driver provisions this automatically. Worker count scales with SF: single-node up to SF=1000; +1 worker per 1000 of SF above that.

Data generation paths

	Spark (default)	Native (DIGen.jar)
Engine	Distributed PySpark	Single-threaded Java
Compute	Serverless	Non-serverless DBR 15.4 + Photon
Output path	…/tpcdi_volume/spark_datagen/sf={SF}/	…/tpcdi_volume/sf={SF}/
File shape	Split (Customer_1.txt, Customer_2.txt, …)	Single (Customer.txt)
Determinism	Same SF → same row counts + audit values	Reference (byte-compatible upstream)
Scaling	Linear across executors	Bound by single-node throughput

The benchmark reads either format via brace-alternation globs ({Customer.txt,Customer_[0-9]*.txt}) so the rest of the pipeline is identical.

Pre-flight on the native path: before any volume side effect, digen_runner verifies (a) java is callable, (b) writable scratch dir exists (/local_disk0 preferred, /tmp fallback for SF≤100), (c) DBR ≤ 15.4. Hard-aborts with a "switch to SINGLE_USER access mode" hint if any check fails — no volume data is touched.

Augmented Incremental — what's different

Standard TPC-DI is heavily skewed to a single bulk historical load — Batch 2 and Batch 3 are tiny by comparison. Augmented Incremental reshapes this into 365 daily increments (2016-07-06 → 2017-07-05), exercising CDC + SCD2 maintenance + cumulative compaction the way a production daily pipeline does.

• Setup DEEP CLONEs the static + dim/fact tables from tpcdi_incremental_staging_{sf} (a shared per-SF staging schema) into a per-user run schema — Liquid layout is inherited directly from staging. Creates per-user _dailybatches/{wh_db}_{sf}/ and _checkpoints/{wh_db}_{sf}/ directories, and emits a 365-day list as a job task value.
• Loop (via Databricks for_each_task): each iteration runs simulate_filedrops (drops one day's pre-staged files into the Autoloader watch dir) → bronze ingest fan-out → silver/gold MERGE incrementals.
• Cleanup is gated by a delete_tables_when_finished condition_task; default is TRUE so the run's schema and Autoloader/checkpoint directories drop on completion. Set the parameter to FALSE per-run if you want to keep the result tables for inspection.
• No audit step — the standard TPC-DI audit checks don't apply to a daily streaming model. (Future work: add row-count parity vs a known-good staged result.)

The SDP variant uses a library-swap trick (update_pipeline_notebook) to bulk-load history with dlt_historical.sql first, then swap to dlt_incremental.sql for the streaming loop. Same physical pipeline, different libraries between phases.

The dbt variant uses Databricks-native dbt_task against a DBSQL warehouse; the dbt project at src/incremental_batches/augmented_incremental/dbt/ is a stock dbt-databricks project (no custom materializations) so it transplants to Snowflake/BigQuery for cross-CDW comparison.

Repo layout (high level)

src/
  TPC-DI Driver.py                            entry-point notebook
  tools/
    data_gen_tasks/data_gen.py                unified entry: digen inline; spark/augmented init+downstream gens
    data_gen_tasks/{gen,copy}_*.py            per-dataset task notebooks
    digen_runner.py                           DIGen.jar wrapper (inline-imported in native mode)
    setup_context.py                          tpcdi_config bootstrap
    workflow_builders/                        Python builders, no Jinja
      datagen_{spark,digen}.py
      workflows_{single_batch,incremental}.py
      sdp_{pipeline,workflow}.py
      augmented_{classic,sdp,dbt,staging}.py
      warehouse.py
    cleanup_after_benchmark.sql               final cleanup task
    tpcdi_gen/                                Spark data-generator modules
      static_audits/sf={sf}/                  pre-computed audit snapshots
    augmented_staging/                        Stage 0 for Augmented Incremental
      _stage_ingestion.py                     stage_to_files() helper
      stage_files/{Dataset}.py                7 partitioned-CSV writers
      cleanup_stage0.py                       drops temp Delta + Batch1/2/3 leftovers
  incremental_batches/                        Cluster + DBSQL benchmark SQL
    bronze/ / silver/ / gold/ / audit_validation/
    augmented_incremental/                    Augmented benchmark (Cluster + SDP + dbt) — see its README
      dbt/                                    dbt variant (its own README)
      DLT/                                    SDP variant (its own README)
  single_batch/
    SQL/                                      single-batch Cluster + DBSQL
    spark_declarative_pipelines/              SDP variant
    dbt/                                      single-batch dbt project (cross-CDW reference)
tests/
  test_workflow_builders.py                   builder unit tests
  smoke_run_workflows.py                      integration smoke
CLAUDE.md                                     architecture context for AI agents

Requirements of Note

• Unity Catalog required for any non-hive_metastore catalog (and for lineage / PK-FK constraints).
• Serverless is optional but assumed for most paths; non-serverless variants need an appropriately sized cluster.
• The native DIGen.jar path requires DBR 15.4 + Photon and a SINGLE_USER cluster access mode for /local_disk0 scratch.
• Augmented Incremental requires Stage 0 (the augmented_staging workflow) to run once per SF before the benchmark. Validated at SF=10/100/1000/5000/10000/20000.
• Augmented Incremental — dbt variant additionally needs a DBSQL warehouse. The Driver auto-creates one sized by SF (Small at SF=20k, doubling SF moves up one size); see the per-SF sizing table above.

Notes on scoring

The TPC-DI scoring metrics haven't been modernized for cloud-native execution, and there has never been an official submission. This implementation does not publish throughput numbers. From the TPC Fair Use Policy:

The Databricks TPC-DI is derived from the TPC-DI and as such is not comparable to published TPC-DI results. The current scoring metrics for the TPC-DI preclude any official submission for cloud-based execution and therefore the Databricks TPC-DI cannot OFFICIALLY be submitted under current scoring metrics. Databricks withholds the ability to submit an official submission to the TPC for this benchmark upon future revision of its scoring metrics. Prior to that, we maintain that this implementation follows all guidelines, rules, and audits required of the official TPC-DI specification.