CI/CD for dbt + Airflow + Spark: Slim CI, State Comparison, Preview Environments

dbt ci cd is the load-bearing engineering discipline that decides whether every pull request to an analytics-engineering monorepo is a 90-second confidence signal or a 45-minute warehouse-burning re-run — and it is the single subsystem senior analytics engineers most often ship half-done, because "we already have GitHub Actions" is not the same as CI that understands manifest.json, deferral, and zero-copy clones. A modern data platform ships three separately-versioned artifacts on every merge — a dbt project, a set of Airflow DAGs, and one or more Spark jobs — each of which mutates warehouse state, and each of which has its own axis of "what does it mean for this PR to be safe to merge?" that generic web-app CI/CD (unit tests + docker build + push) simply does not model. The engineering trade-off does not live in "should we run CI on data pipelines" — every team past two engineers needs it — but in how you scope each run so the fast path is fast, the slow path is honest, and the warehouse bill in November does not triple because CI decided to rebuild the entire mart on every whitespace-only PR.

This guide is the senior data-platform walkthrough you wished existed the first time an interviewer asked "walk me through slim CI for dbt and why state:modified+ matters", or "your PR only edits one model — why did the CI job just spend $180 on Snowflake compute?", or "how do you give analysts a preview URL for the dashboard they're editing without cloning the entire warehouse?". It covers the four canonical CI axes (correctness, cost, latency-to-merge, blast radius), the slim CI recipe with dbt build --select state:modified+ --defer --state prod-manifest/ plus dbt clone for zero-copy previews, the Airflow DAG-parse gate + operator unit-test matrix that catches import-time breaks before the scheduler ever sees them, the Spark packaged-JAR + local pytest + dry-run submit sequence, and the preview-environment pattern where every open pull request maps to its own pr_<PR>_<branch> schema on Snowflake or Databricks Unity Catalog that is auto-created on PR open and auto-dropped on PR close. Each section pairs a teaching block with a Solution-Tail interview answer — code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the ETL practice library →, rehearse on the SQL practice library →, and sharpen the systems axis with the design practice library →.

---

On this page

• Why generic CI/CD breaks on data pipelines
• Slim CI for dbt — state:modified, defer, and clone
• Airflow CI — DAG parse, integration tests, deployable artifacts
• Spark CI — packaged jobs, unit + integration tests, dry-run submits
• Preview environments — one branch, one warehouse schema
• Cheat sheet — CI/CD recipes for dbt + Airflow + Spark
• Frequently asked questions
• Practice on PipeCode

---

1. Why generic CI/CD breaks on data pipelines — the four axes

Four systems, four different notions of "safe to merge" — the invariant every senior architect encodes

The one-sentence invariant: ci for data pipelines is a picking exercise across four systems (dbt, Airflow, Spark, and the warehouse itself), each with its own definition of "the PR is safe" — dbt needs state:modified+ deferral so the CI job rebuilds only changed models against parent tables in prod, Airflow needs a DAG-parse gate so import-time errors never reach the scheduler, Spark needs a packaged job + local unit tests + a dry-run submit so the cluster is only touched after config is validated, and the warehouse needs a per-PR preview schema so blast radius stays bounded — and shipping any of the four without its native gate turns CI into either "45-minute full rebuild theatre" or "green build, broken prod". The default you pick in month one becomes the default you fight to migrate away from in year three, because every analyst dashboard, every Airflow DAG, and every Spark job hard-codes assumptions about where the "safe to merge" line lives.

The four axes generic CI/CD misses.

• Correctness. Web-app CI answers "did the unit tests pass?" Data-pipeline CI has to answer and "does the new dbt model actually produce the same PK count as the old one?", "does the DAG still parse under the current Airflow version?", "does the Spark job's schema still match the upstream Kafka topic?". Each system's correctness gate is different; encoding all three is the senior-signal answer.
• Cost. A full dbt build against Snowflake for a 200-model warehouse can cost $50-$500 per CI run. Multiply by 30 PRs/day and the CI compute bill dwarfs the actual production bill. Slim CI + preview schemas + XS warehouses are how you keep the November invoice under control. Interviewers open with this question because it separates people who have paid the bill from those who have not.
• Latency-to-merge. A CI run that takes 45 minutes turns PR review into a next-day cycle. A CI run that takes 90 seconds keeps analysts in flow. The dbt slim-CI + Airflow parse-gate + Spark local-unit-test sequence is designed for the 90-second floor; skipping any layer pushes latency back into the 20-45 minute regime.
• Blast radius. Web-app CI writes to a docker layer cache and a test bucket. Data-pipeline CI writes to actual warehouse schemas — if you don't isolate per PR, one CI job's dbt seed overwrites another CI job's fixture and both fail. The preview-schema pattern gives every PR its own pr_<PR>_<branch> namespace so blast radius is exactly one PR.

The 2026 reality — slim CI everywhere, state comparison, preview environments.

• Slim CI is the default for dbt. dbt build --select state:modified+ --defer --state prod-manifest/ is the four-word answer. Every senior interview lists it in the first minute. Rebuilding only changed models against prod parents cuts CI from 45 minutes to 2 minutes on a typical monorepo.
• State comparison is the default for artifact promotion. The manifest.json from the last prod run is downloaded as an artifact; the CI job compares its own manifest against it and computes the state:modified set. Without prod manifest, slim CI does not work.
• Preview environments are the default for warehouse UX. Every PR gets a pr_<PR>_<branch> schema (Snowflake CREATE SCHEMA ... CLONE, Databricks CREATE SCHEMA ... CLONE). Analysts hit the preview URL; when the PR merges or closes, the schema is dropped. Zero-copy clones make this cheap.
• Airflow parse gates are non-negotiable. A DAG that fails at import time takes down every DAG in the DagBag. The parse gate is a one-line CI check that catches it before the scheduler ever tries to load it.
• Spark dry-run submits are the last-mile gate. spark-submit --deploy-mode client --num-executors 0 (or the equivalent Databricks JOBS API dry-run) validates config, JAR resolution, and cluster reachability without actually spinning up executors — the cheapest possible "does this job actually work in this cluster" gate.

What interviewers listen for.

• Do you name all four systems (dbt, Airflow, Spark, warehouse) and their per-system gates without prompting? — senior signal.
• Do you say "slim CI with state:modified+ and deferral" in the first minute when dbt CI comes up? — required answer.
• Do you push back on "just run the full test suite" with the cost argument — "that's a $180 CI run per PR"? — senior signal.
• Do you name the preview-environment pattern as the answer to "how do analysts review a PR"? — senior signal.
• Do you describe CI as "gates that shrink blast radius" rather than as vague "running tests"? — required answer.

Worked example — the four-axis comparison table across dbt, Airflow, Spark

Detailed explanation. The single most useful artifact for a data-pipeline CI/CD interview is a memorised 4×3 comparison table: four axes (correctness, cost, latency, blast radius) crossed with three systems (dbt, Airflow, Spark). Every senior CI/CD discussion converges on this table within the first ten minutes; having it in your head is what separates a fluent answer from a stumbling one. Walk through building the table for a hypothetical analytics-engineering monorepo.

• Repo shape. One monorepo containing dbt/, airflow/dags/, spark/jobs/. Deploys to Snowflake (warehouse), MWAA (Airflow), and EMR-on-EKS (Spark).
• PR volume. ~30 pull requests per day; ~5 touch dbt, ~5 touch Airflow, ~2 touch Spark, the rest touch shared libraries.
• Target CI budget. Under 5 minutes p95 for a dbt-only PR; under 8 minutes for a mixed PR; under $10 per CI run.

Question. Build the four-axis CI comparison for dbt, Airflow, and Spark and pick the gate each system needs.

Input.

Axis	dbt	Airflow	Spark
Correctness gate	state:modified+ rebuild + dbt test	DAG parse + operator unit tests	pytest with local SparkSession
Cost gate	slim CI + XS warehouse	container reuse + shared runner	local mode + --num-executors 0 dry-run
Latency gate	2 min for slim, 45 min for full	30 s parse, 3 min unit tests	4 min local pytest, 30 s dry-run
Blast radius gate	pr_<PR>_<branch> schema	separate MWAA environment	temp S3 prefix + separate EMR cluster

Code.

# .github/workflows/data-pipeline-ci.yml
name: data-pipeline-ci
on:
  pull_request:
    branches: [ main ]

jobs:
  detect-changes:
    runs-on: ubuntu-latest
    outputs:
      dbt:     ${{ steps.filter.outputs.dbt }}
      airflow: ${{ steps.filter.outputs.airflow }}
      spark:   ${{ steps.filter.outputs.spark }}
    steps:
      - uses: actions/checkout@v4
      - uses: dorny/paths-filter@v3
        id: filter
        with:
          filters: |
            dbt:
              - 'dbt/**'
            airflow:
              - 'airflow/**'
            spark:
              - 'spark/**'

  dbt-slim-ci:
    needs: detect-changes
    if: needs.detect-changes.outputs.dbt == 'true'
    uses: ./.github/workflows/dbt-slim.yml

  airflow-parse-and-test:
    needs: detect-changes
    if: needs.detect-changes.outputs.airflow == 'true'
    uses: ./.github/workflows/airflow.yml

  spark-unit-and-dry-run:
    needs: detect-changes
    if: needs.detect-changes.outputs.spark == 'true'
    uses: ./.github/workflows/spark.yml

Step-by-step explanation.

1. The top-level workflow gates every downstream job on a paths-filter step so a whitespace-only PR to README.md does not trigger a dbt build. Path-based short-circuiting is the cheapest CI optimisation and pays for itself the first day.
2. Each system gets its own reusable workflow file. Isolating dbt, Airflow, and Spark into separate files keeps the top-level data-pipeline-ci.yml readable and lets each team own its own gate logic without cross-editing.
3. The dbt workflow (covered in section 2) runs slim CI against a preview schema. It downloads the prod manifest.json, computes state:modified+, defers parents to prod, and rebuilds only what changed.
4. The Airflow workflow (covered in section 3) runs the DAG parse gate first (fastest fail), then the operator unit tests, then optionally an integration test against a LocalExecutor.
5. The Spark workflow (covered in section 4) runs local unit tests with pyspark.testing, packages the job into a wheel or JAR, and runs a dry-run submit against a small dev cluster to validate config.

Output.

PR touches	Jobs that run	p95 CI time	Approximate cost
dbt only	dbt-slim-ci	2 min	$0.30
Airflow only	airflow-parse-and-test	3 min	$0.05
Spark only	spark-unit-and-dry-run	4 min	$0.08
Mixed (dbt + Airflow)	both	4 min (parallel)	$0.35
README only	none	0 s	$0.00

Rule of thumb. Never run a dbt build on a PR that does not touch dbt. Path-based gating is the single most impactful CI optimisation for a data-pipeline monorepo. Combine it with slim CI, DAG parse gates, and Spark dry-runs and a mixed-PR run finishes in under 5 minutes.

Worked example — what interviewers actually probe on data-pipeline CI/CD

Detailed explanation. The senior data-platform CI/CD interview has a predictable structure: the interviewer opens with an ambiguous question ("how do you set up CI for a dbt project?"), then progressively narrows to test whether you know the axes. The candidates who name slim CI and state comparison in sentence one score highest; the candidates who describe "GitHub Actions running dbt run" score lowest. Walk through the interview grading rubric.

• Ambiguous opener. "How would you set up CI for our dbt project?" — invites you to name slim CI.
• Follow-up 1. "What happens when the PR only edits one model?" — probes state:modified.
• Follow-up 2. "How do you avoid rebuilding all parent tables?" — probes deferral.
• Follow-up 3. "How much does a CI run cost?" — probes cost axis.
• Follow-up 4. "How do analysts review the changed dashboard?" — probes preview environments.
• Follow-up 5. "What about Airflow and Spark in the same repo?" — probes multi-system gates.

Question. Draft a 5-minute senior CI/CD answer that covers all three systems and all four axes without waiting to be asked.

Input.

Interview signal	Weak answer	Senior answer
dbt gate named	"we run dbt run in CI"	"slim CI with dbt build --select state:modified+ --defer --state prod-manifest/"
Airflow gate named	"we run pytest"	"DAG parse gate first, then operator unit tests, then integration test with LocalExecutor"
Spark gate named	"we submit the job"	"local pytest + packaged JAR + dry-run submit against dev cluster"
Cost bound	"not sure"	"under $1/run, XS warehouse with AUTO_SUSPEND=60"
Preview UX	"analysts wait for merge"	"pr_<PR>_<branch> schema created on PR open, dropped on close"

Code.

Senior data-pipeline CI/CD answer template (5 minutes)
======================================================

Minute 1 — name the gates up front
  "For each of the three systems in a data monorepo, CI has a native
   gate. dbt: slim CI with state:modified+ and deferral. Airflow:
   DAG parse gate + operator unit tests. Spark: local pytest +
   dry-run submit."

Minute 2 — dbt slim CI
  "The command is: dbt build --select state:modified+ --defer
   --state prod-manifest/ --target ci. The prod manifest.json is
   downloaded from S3 (uploaded on every prod run). state:modified+
   picks the changed models and their descendants; --defer routes
   any unbuilt parent refs to the prod schema instead of the CI
   schema. This turns a 45-minute full rebuild into a 2-minute
   incremental."

Minute 3 — Airflow CI
  "First gate is a DagBag import: python -c 'from airflow.models
   import DagBag; assert not DagBag(dag_folder=\"airflow/dags\",
   include_examples=False).import_errors'. Fails in 15 seconds if
   any DAG file breaks at import. Second gate is unit tests on
   custom operators and hooks. Third gate is an integration test
   with LocalExecutor for the DAGs that touch shared resources."

Minute 4 — Spark CI + preview envs
  "Spark: pytest against a local SparkSession fixture for logic,
   docker-compose Spark mini-cluster for schema and shuffle behaviour,
   and spark-submit --deploy-mode client --num-executors 0 as a
   dry-run gate that validates cluster config without spinning up
   executors. Preview envs: every PR gets a pr_<PR>_<branch> schema
   via Snowflake CREATE SCHEMA ... CLONE, with AUTO_SUSPEND=60 on the
   compute; dropped on PR close via GitHub webhook."

Minute 5 — cost + blast radius
  "Cost: slim CI + XS warehouse + AUTO_SUSPEND=60 keeps a dbt CI run
   under $0.30 typically. Blast radius: per-PR schema means one CI
   job cannot corrupt another. On PR close, the cleanup workflow
   drops the schema and de-registers the compute. November invoice
   is bounded because ephemeral compute suspends within 60 s of the
   last query."

Step-by-step explanation.

1. Minute 1 is the crucial framing. Naming the three system-specific gates immediately signals you understand the axes; weak candidates dive into a single tool ("we use GitHub Actions and…") before naming any gate.
2. Minute 2 addresses the dbt cost axis with the exact state:modified+ --defer command. This preempts the "how do you avoid rebuilding all parents?" follow-up by answering it before it is asked.
3. Minute 3 addresses the Airflow correctness axis: the DAG parse gate is the fast-fail; the operator unit tests are the coverage layer; the integration test is the "does the whole DAG wire together?" layer. Naming all three shows you have shipped Airflow CI, not just read the docs.
4. Minute 4 combines Spark and preview environments into one minute because they share a design theme: cheap ephemeral compute that suspends when idle. --num-executors 0 for Spark and AUTO_SUSPEND=60 for Snowflake are both "compute suspends when unused" patterns.
5. Minute 5 covers the cost + blast-radius axes explicitly. Naming the November invoice concern signals you have paid the bill; naming the auto-cleanup on PR close signals you have on-called the workflow.

Output.

Grading criterion	Weak score	Senior score
Names all four axes	rare	mandatory
Names slim CI in minute 1	rare	required
Names DAG parse gate	occasional	mandatory
Names dry-run Spark submit	rare	senior signal
Names preview env + cleanup	rare	senior signal
Names cost budget in dollars	rare	senior signal

Rule of thumb. The senior data-pipeline CI/CD answer is a 5-minute monologue that covers dbt, Airflow, and Spark plus the four axes (correctness, cost, latency, blast radius) without waiting for the follow-ups. Rehearse it once; deploy it every time.

Worked example — the pick-the-gate decision tree per PR type

Detailed explanation. Given a new pull request, the senior architect runs a fast decision tree in their head to pick which gates to run. Codifying the tree makes the CI system answer reproducible: any change fires only the gates that could possibly detect a regression from that change, and skips the rest. Walk through the tree for four canonical PR types.

• Q1. Which files did the PR touch? → path-filter output.
• Q2. If dbt files → does the PR add/modify a .sql model, a .yml schema test, a macro, or a seed? → each triggers different subsets of slim CI.
• Q3. If Airflow files → is it a DAG, an operator, a hook, or a plugin? → each has a different test target.
• Q4. If Spark files → is it a job entry point, a UDF, or shared library code? → each has different local + dry-run coverage.
• Q5. If shared libraries → run every downstream system's smoke test.

Question. Walk the decision tree for four scenarios and record the gates that fire.

Input.

Scenario	Files touched	Q2/3/4	Gates that fire
Edit one dbt model	dbt/models/marts/orders.sql	model	slim CI (state:modified+), dbt test
Add an Airflow operator	airflow/plugins/operators/reverse_etl.py	operator	operator unit tests, DAG parse
Fix a Spark UDF	spark/jobs/enrich/udf/geo.py	UDF	local pytest, dry-run submit
Bump a shared library	shared/pyutils/dates.py	shared	all three systems' smoke tests

Code.

# scripts/pick_ci_gates.py — illustrative
def pick_gates(files_touched: list[str]) -> set[str]:
    """Return the set of CI gates that must run for a given file diff."""
    gates: set[str] = set()

    if any(f.startswith("dbt/") for f in files_touched):
        gates |= {"dbt-slim-ci", "dbt-test", "preview-schema"}

    if any(f.startswith("airflow/") for f in files_touched):
        gates |= {"airflow-parse", "airflow-unit"}
        if any("dags/" in f for f in files_touched):
            gates.add("airflow-integration")

    if any(f.startswith("spark/") for f in files_touched):
        gates |= {"spark-unit"}
        if any("jobs/" in f for f in files_touched):
            gates |= {"spark-package", "spark-dry-run"}

    if any(f.startswith("shared/") for f in files_touched):
        # Shared code — smoke-test every downstream system
        gates |= {"dbt-slim-ci", "airflow-parse", "spark-unit"}

    return gates


# Walk the four scenarios
print(pick_gates(["dbt/models/marts/orders.sql"]))
# → {'dbt-slim-ci', 'dbt-test', 'preview-schema'}

print(pick_gates(["airflow/plugins/operators/reverse_etl.py"]))
# → {'airflow-parse', 'airflow-unit'}

print(pick_gates(["spark/jobs/enrich/udf/geo.py"]))
# → {'spark-unit', 'spark-package', 'spark-dry-run'}

print(pick_gates(["shared/pyutils/dates.py"]))
# → {'dbt-slim-ci', 'airflow-parse', 'spark-unit'}

Step-by-step explanation.

1. Scenario 1 — a single dbt model edit. Slim CI computes state:modified+, rebuilds orders and its descendants against the preview schema, runs the model's .yml tests, and posts pass/fail to the PR. Total: ~2 minutes.
2. Scenario 2 — an operator edit. The operator is not imported by any DAG at file-scan time, so the parse gate is fast (~15 s). The operator unit tests exercise the operator directly with a mock hook. No integration test fires because no DAG file changed.
3. Scenario 3 — a Spark UDF edit. Local pytest exercises the UDF against a SparkSession.builder.master("local[2]") fixture. The job entry point that uses the UDF is repackaged and dry-run submitted to validate the cluster still accepts it.
4. Scenario 4 — a shared library change. Because we don't know which downstream systems use pyutils.dates, we run one smoke gate for each: dbt slim CI (which will trip if any macro imports the library), Airflow parse (which will trip if any DAG imports it), Spark unit tests (which will trip if any job imports it).
5. The decision tree is not a heuristic — it is a contract. Every gate in the tree has to run when the tree says it should; skipping any gate breaks the invariant that CI-green means merge-safe.

Output.

Scenario	Gates fired	p95 latency
Single dbt model edit	slim CI + dbt test + preview schema	2 min
New operator	parse + operator unit	30 s
Spark UDF fix	unit + package + dry-run	4 min
Shared library bump	slim CI + parse + spark unit (parallel)	4 min

Rule of thumb. The four-question decision tree is a whiteboard-friendly answer. Practice walking it end-to-end so an interviewer can hand you any PR shape and get a gate list in under 30 seconds.

Senior interview question on data-pipeline CI/CD architecture

A senior interviewer often opens with: "You inherit a data-pipeline monorepo (dbt + Airflow + Spark) whose CI takes 45 minutes and costs $180 per PR. Walk me through the redesign — the per-system gates, the state comparison, the preview environments, the cost budget, and the on-call runbook when CI is red on prod-blocking PRs."

Solution Using slim CI + parse gate + dry-run submit + per-PR preview schemas

# .github/workflows/pipeline-ci.yml — the whole redesign
name: pipeline-ci
on:
  pull_request:
    branches: [ main ]
  pull_request_target:
    types: [ closed ]

jobs:
  # 1. Path-based short-circuit — never run a system that did not change
  detect:
    runs-on: ubuntu-latest
    outputs:
      dbt:     ${{ steps.f.outputs.dbt }}
      airflow: ${{ steps.f.outputs.airflow }}
      spark:   ${{ steps.f.outputs.spark }}
    steps:
      - uses: actions/checkout@v4
      - id: f
        uses: dorny/paths-filter@v3
        with:
          filters: |
            dbt:     [ 'dbt/**' ]
            airflow: [ 'airflow/**' ]
            spark:   [ 'spark/**' ]

  # 2. Preview schema — created per PR, dropped on close (see section 5)
  preview-schema:
    needs: detect
    if: needs.detect.outputs.dbt == 'true' && github.event.action != 'closed'
    uses: ./.github/workflows/preview-schema-create.yml

  # 3. dbt slim CI (see section 2)
  dbt-ci:
    needs: [ detect, preview-schema ]
    if: needs.detect.outputs.dbt == 'true'
    uses: ./.github/workflows/dbt-slim.yml

  # 4. Airflow gates (see section 3)
  airflow-ci:
    needs: detect
    if: needs.detect.outputs.airflow == 'true'
    uses: ./.github/workflows/airflow.yml

  # 5. Spark gates (see section 4)
  spark-ci:
    needs: detect
    if: needs.detect.outputs.spark == 'true'
    uses: ./.github/workflows/spark.yml

  # 6. On PR close — clean up preview schema regardless of merge/reject
  cleanup:
    if: github.event.action == 'closed'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Drop preview schema
        run: python scripts/drop_preview_schema.py --pr ${{ github.event.number }}

# CI budget contract — one line per gate
Gate                    p95 latency   Cost budget   Owner
────────────────────────────────────────────────────────────
dbt-slim-ci             120 s         $0.30         analytics-eng
airflow-parse           15 s          $0.02         platform
airflow-unit            180 s         $0.05         platform
spark-unit              240 s         $0.08         data-eng
spark-dry-run           30 s          $0.02         data-eng
preview-schema-create   20 s          $0.01         platform
preview-schema-drop     10 s          $0.00         platform
────────────────────────────────────────────────────────────
Mixed PR (all systems)  ~5 min        ~$0.50        —

Step-by-step trace.

Concern	Before	After
dbt strategy	full dbt run --target ci	dbt build --select state:modified+ --defer
CI duration (dbt-only PR)	45 min	2 min
CI cost (dbt-only PR)	$180	$0.30
Airflow gate	none	parse gate + operator unit tests
Spark gate	submit-to-prod-cluster	local pytest + dry-run submit
Warehouse blast radius	shared analytics_ci schema	per-PR pr_<PR>_<branch> schema
Preview UX for analysts	wait for merge	preview URL on PR open
Cleanup	manual quarterly purge	auto-drop on PR close

After the migration, a dbt-only PR completes CI in about 2 minutes and costs roughly $0.30; a mixed PR touching all three systems completes in about 5 minutes because gates run in parallel; the November warehouse invoice drops from about $9,000 in CI compute to about $200. Analysts open the preview URL from the PR description and see the dashboard rendered against the changed models within 3 minutes of PR open.

Output:

Metric	Before	After
Median CI time	45 min	2 min
p95 CI time (mixed PR)	55 min	5 min
Median CI cost	$180	$0.30
Preview environment	none	one per PR
Blast radius	one shared schema	per-PR isolation
November warehouse CI bill	~$9,000	~$200

Why this works — concept by concept:

• Path-based short-circuit — the cheapest optimisation available. A whitespace-only PR to README.md fires zero gates; the CI pipeline is essentially free for docs-only work. This is the "detect" job at the top of the workflow.
• Per-system gates — dbt, Airflow, and Spark each get the native gate their runtime actually cares about. Slim CI for dbt, parse gate for Airflow, dry-run submit for Spark. Running only the gates that could possibly detect a regression is the definition of proportionate CI.
• Preview schemas — every PR gets its own pr_<PR>_<branch> schema, so two concurrent CI runs cannot collide. The preview is also the analyst-facing URL; the same schema serves CI and human review.
• Auto-cleanup on PR close — the cleanup workflow is idempotent and fires whether the PR merges or is rejected. Without it, preview schemas accumulate indefinitely; with it, warehouse footprint is bounded.
• Cost budget per gate — publishing the cost budget makes it a contract, not a hope. When a gate exceeds its budget, the SRE-of-CI catches it in the weekly review; when a new gate is proposed, the budget question comes up in code review.
• Cost — one CI orchestration layer, one preview-schema helper, one cleanup workflow, plus per-system gates that each stay under $0.30. Compared to the 45-minute full-rebuild baseline, this is one order of magnitude faster at less than 1% of the cost. Net O(delta) per PR versus O(catalog) per PR.

ETL
Topic — etl
ETL problems on pipeline-CI design

Practice →

Design Topic — design Design problems on data platform CI/CD

Practice →

---

2. Slim CI for dbt — state:modified, defer, and clone

dbt build --select state:modified+ --defer --state prod-manifest/ is the four-word answer to every dbt CI/CD interview

The mental model in one line: slim ci dbt is the pattern where a CI job downloads the last successful production manifest.json, compares its own manifest against it to compute state:modified (the set of models whose SQL, config, or upstream sources actually changed), rebuilds only that set plus its descendants (state:modified+) into a per-PR schema, and defers any unbuilt parent refs to the production schema instead of re-materialising them — turning a 45-minute full rebuild into a 2-minute incremental that costs cents instead of dollars while still testing every change against real prod data. Every senior analytics engineer has adopted this pattern once; every senior interviewer probes for it in the first minute.

The four axes for slim CI dbt.

• Correctness. state:modified catches every model whose SQL, config, or upstream .yml schema actually changed. The + selector suffix adds every descendant (models that ref the changed ones), so a change to a staging model rebuilds the mart that depends on it. This is the correctness contract: if the PR could affect a model, that model gets rebuilt and tested.
• Cost. A slim CI run rebuilds 3-15 models on average (versus 200-1500 for a full run). Snowflake XS warehouse with AUTO_SUSPEND=60 keeps a typical run under $0.30. This is the cost contract; without slim CI, a 200-model warehouse hits $30-$500 per CI run.
• Latency-to-merge. Two-minute CI turns "I'll review this after lunch" into "I'll review this in the next 5 minutes." The latency payoff compounds across a team of ten analysts each opening 3 PRs a day.
• Blast radius. Slim CI writes to the per-PR preview schema (pr_<PR>_<branch>), never to analytics_prod. Deferral routes reads to the prod schema (safe — read-only) and writes to the preview schema (isolated — per-PR). The prod schema is never touched by CI.

The manifest.json — the state comparison substrate.

• What it is. dbt's target/manifest.json describes the parsed project: every model, its SQL fingerprint, its config, its upstream refs, its sources. Two manifests are directly comparable; state:modified diffs them.
• Where it lives. Uploaded to S3 (or GCS or Azure Blob) as an artifact at the end of every successful prod run. Path convention: s3://dbt-artifacts/prod/latest/manifest.json plus a versioned copy at s3://dbt-artifacts/prod/<run_id>/manifest.json.
• How CI downloads it. The CI job pulls the "prod latest" manifest into a local prod-manifest/ directory before running slim CI. The --state prod-manifest/ flag points dbt at it.
• The bootstrap problem. The first CI run has no prod manifest to compare against; the slim-CI logic falls back to a full build. Most teams solve this by seeding the S3 path from a manual initial run.

The state:modified+ selector — what it catches and what it doesn't.

• What state:modified catches. Models whose SQL text changed, whose config (materialization, tags, cluster_by, etc.) changed, whose upstream sources' definitions changed, or whose upstream refs' fingerprints changed. Also new models and deleted models.
• What + adds. Every descendant (children, grandchildren, etc.) of the modified set. If you change a staging model, every mart that depends on it rebuilds.
• What it does not catch. Behaviour changes in seeds (use state:modified,resource_type:seed for that), macro changes (use state:modified.macros), or changes in a dbt package version bump (use state:modified.macros + state:modified.contracts).
• The safety escape hatch. For a suspicious PR (a big refactor, a dbt-core upgrade), append --select "state:modified+" "state:new+" or fall back to a full build on the main branch periodically.

The --defer --state prod-manifest/ pair — the parent-routing magic.

• The read. When a CI-built model runs {{ ref('unmodified_upstream') }}, dbt sees that model was not rebuilt in this run, checks the state manifest, and rewrites the ref to point at the prod schema instead of the CI schema.
• The write. All modified models still write to the CI schema (pr_<PR>_<branch>); deferred parents are only read from prod. Prod is never mutated.
• The permission model. The CI role needs SELECT on the prod schema (to read deferred parents) and CREATE + SELECT + INSERT on the preview schema (to build modified models). Configure once in Snowflake / BigQuery / Redshift; use forever.

Common interview probes on slim CI dbt.

• "What does state:modified+ mean?" — required answer: modified models plus all descendants.
• "How do you get the prod manifest into CI?" — download from S3 at the top of the CI job.
• "What does --defer do?" — routes unbuilt refs to the state manifest's schema (prod) instead of the CI schema.
• "How do you handle the bootstrap case where no prod manifest exists yet?" — fall back to a full build on the first run and seed the artifact.
• "How do you avoid slim CI missing a macro change?" — either use state:modified.macros selector or trigger a full build on macro-file changes.

Worked example — GitHub Actions job that runs slim CI on every PR

Detailed explanation. The canonical slim CI setup: a GitHub Actions workflow that installs dbt, downloads the prod manifest from S3, computes state:modified+, defers parents to prod, and posts the results to the PR. Build the whole thing from scratch.

• Trigger. pull_request open, synchronize, reopen; skip on PR close.
• Runner. ubuntu-latest; Python 3.11; dbt-snowflake 1.9+.
• State source. Prod manifest from s3://dbt-artifacts/prod/latest/manifest.json.
• Target schema. pr_<PR>_<branch> derived from GitHub context.
• Post. Slim-CI summary as a PR comment.

Question. Write the full GitHub Actions workflow that runs dbt build --select state:modified+ --defer --state prod-manifest/ on every PR.

Input.

Parameter	Value
CI runner	ubuntu-latest
dbt adapter	dbt-snowflake 1.9
State source	s3://dbt-artifacts/prod/latest/
Target	ci (pr__ schema)
Auth	OIDC to AWS + Snowflake key-pair from GitHub secret

Code.

# .github/workflows/dbt-slim.yml — reusable slim CI workflow
name: dbt-slim-ci
on:
  workflow_call:

jobs:
  slim-ci:
    runs-on: ubuntu-latest
    permissions:
      id-token: write          # OIDC to AWS
      contents: read
      pull-requests: write     # to post the comment
    env:
      DBT_PROFILES_DIR: ./ci
      DBT_TARGET: ci
      DBT_PR_NUMBER: ${{ github.event.pull_request.number }}
      DBT_BRANCH_SLUG: ${{ github.head_ref }}

    steps:
      - name: Check out PR
        uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install dbt
        working-directory: dbt
        run: |
          pip install -r requirements.txt
          dbt --version

      - name: Configure AWS credentials (OIDC)
        uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: arn:aws:iam::123456789012:role/gha-dbt-ci
          aws-region: us-east-1

      - name: Download prod manifest
        run: |
          mkdir -p prod-manifest
          aws s3 cp s3://dbt-artifacts/prod/latest/manifest.json \
                    prod-manifest/manifest.json

      - name: Compute preview schema name
        id: schema
        run: |
          # pr_<PR>_<branch-slug> — sanitize branch name for SQL identifier
          slug=$(echo "$DBT_BRANCH_SLUG" | tr '/' '_' | tr -cd '[:alnum:]_' | tr '[:upper:]' '[:lower:]' | cut -c1-40)
          echo "schema=pr_${DBT_PR_NUMBER}_${slug}" >> "$GITHUB_OUTPUT"

      - name: dbt deps
        working-directory: dbt
        run: dbt deps

      - name: dbt build (slim CI)
        working-directory: dbt
        env:
          DBT_SNOWFLAKE_PRIVATE_KEY: ${{ secrets.SNOWFLAKE_PRIVATE_KEY }}
          DBT_TARGET_SCHEMA: ${{ steps.schema.outputs.schema }}
        run: |
          dbt build \
            --select state:modified+ \
            --defer \
            --state ../prod-manifest \
            --target ci \
            --fail-fast

      - name: Upload CI manifest for downstream jobs
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: ci-manifest-${{ env.DBT_PR_NUMBER }}
          path: dbt/target/manifest.json

      - name: Post slim-CI summary to PR
        if: always()
        uses: actions/github-script@v7
        with:
          script: |
            const fs   = require('fs');
            const path = require('path');
            const run  = fs.readFileSync('dbt/target/run_results.json', 'utf8');
            const j    = JSON.parse(run);
            const rows = j.results.map(r =>
              `| ${r.unique_id} | ${r.status} | ${r.execution_time.toFixed(2)}s |`
            ).join('\n');
            const body = `### dbt slim CI — PR #${{ env.DBT_PR_NUMBER }}\n\n` +
              `| Model | Status | Time |\n|---|---|---|\n${rows}\n\n` +
              `Preview schema: \`${{ steps.schema.outputs.schema }}\`.\n`;
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner:  context.repo.owner,
              repo:   context.repo.repo,
              body:   body
            });

# dbt/ci/profiles.yml — CI target profile
analytics:
  target: dev
  outputs:
    ci:
      type: snowflake
      account: "{{ env_var('SNOWFLAKE_ACCOUNT') }}"
      user: "{{ env_var('SNOWFLAKE_USER') }}"
      private_key: "{{ env_var('DBT_SNOWFLAKE_PRIVATE_KEY') }}"
      role: DBT_CI_ROLE
      warehouse: DBT_CI_WH_XS   # AUTO_SUSPEND=60
      database: ANALYTICS
      schema: "{{ env_var('DBT_TARGET_SCHEMA') }}"  # pr_<PR>_<branch>
      threads: 8
      client_session_keep_alive: false

Step-by-step explanation.

1. The workflow is workflow_call-able so the top-level pipeline-ci.yml can invoke it as a reusable job. This keeps the dbt-specific logic in one file and lets the top-level orchestrator focus on gate selection.
2. OIDC + aws-actions/configure-aws-credentials is the modern auth pattern; there is no long-lived AWS key in a GitHub secret. The IAM role gha-dbt-ci has read access to the dbt-artifacts bucket only.
3. Compute preview schema name derives pr_<PR>_<slug> from the GitHub Actions context. The slug sanitiser strips non-alphanumeric characters, lowercases, and caps at 40 characters — Snowflake identifiers must fit in 255 characters and be lowercase-safe.
4. dbt build --select state:modified+ --defer --state ../prod-manifest --target ci is the core command. build is run + test + snapshot + seed in dependency order — one command runs everything. --fail-fast stops on the first error so CI feedback is fast.
5. The final step posts a table of every rebuilt model + its status + its runtime as a PR comment. Reviewers see the exact impact of the PR without opening the CI logs. The comment updates on every push (or a new comment; use a comment-updater action for one-comment-per-PR UX).

Output.

Rebuild scope	Full CI	Slim CI
Models rebuilt (median PR)	200	4
p95 CI time	45 min	2 min
p95 CI cost	$180	$0.30
Preview schema	analytics_ci (shared)	pr_<PR>_<branch> (isolated)
PR comment	none	model-by-model table

Rule of thumb. For any dbt CI setup, mount slim CI as the default gate: state:modified+ --defer --state prod-manifest/. Post the results as a PR comment. Upload the CI manifest so downstream jobs (e.g. a doc-preview or dbt-freshness job) can consume it. These three habits move a dbt project from "CI is a chore" to "CI is a review tool."

Worked example — state comparison against manifest.json from main

Detailed explanation. State comparison is the substrate slim CI runs on: dbt compares two manifests and computes the changed set. Walk through what "state comparison" actually means at the manifest level — which fields dbt looks at, how it computes state:modified, and how you can inspect the comparison manually to debug false positives.

• Fields dbt fingerprints. SQL text (after Jinja render), config (materialization, cluster_by, partition_by, tags, meta), upstream refs, upstream sources, contracts (if enabled), and constraints.
• The fingerprint algorithm. dbt computes a SHA-256 hash of the concatenation of the above fields per model. Two models with the same hash are "unmodified"; differing hashes mark state:modified.
• Why false positives happen. Whitespace changes in SQL, Jinja variable resolution differences (e.g. {{ var('env') }} resolves differently in CI vs prod), or macro upgrades can flip the hash even when the effective SQL is identical.
• Debugging. dbt ls --select state:modified --state prod-manifest/ lists exactly what changed; dbt-checkpoint or the dbt-project-evaluator package can help diagnose surprising diffs.

Question. Write a debug script that lists exactly which fields of a modified model changed between the CI manifest and the prod manifest.

Input.

Component	Value
Prod manifest	prod-manifest/manifest.json
CI manifest	dbt/target/manifest.json
Model to inspect	model.analytics.orders_mart
Fields to diff	raw_code, config.materialized, depends_on.nodes, checksum.checksum

Code.

# scripts/dbt_state_diff.py — diff two manifests for a single model
"""
Usage:
  python scripts/dbt_state_diff.py \
      --prod prod-manifest/manifest.json \
      --ci   dbt/target/manifest.json \
      --model model.analytics.orders_mart
"""
import argparse
import json
import difflib
from pathlib import Path


def load_node(path: Path, unique_id: str) -> dict:
    manifest = json.loads(path.read_text())
    node = manifest["nodes"].get(unique_id)
    if node is None:
        raise SystemExit(f"{unique_id!r} not in {path}")
    return node


def diff_fields(prod: dict, ci: dict) -> list[tuple[str, str, str]]:
    """Return (field, prod_value, ci_value) for each differing top-level field."""
    fields = [
        "raw_code",
        "checksum",
        "config",
        "depends_on",
        "sources",
        "refs",
        "contract",
    ]
    out: list[tuple[str, str, str]] = []
    for f in fields:
        pv = json.dumps(prod.get(f), sort_keys=True, indent=2)
        cv = json.dumps(ci.get(f),  sort_keys=True, indent=2)
        if pv != cv:
            out.append((f, pv, cv))
    return out


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--prod",  required=True, type=Path)
    ap.add_argument("--ci",    required=True, type=Path)
    ap.add_argument("--model", required=True)
    args = ap.parse_args()

    prod = load_node(args.prod, args.model)
    ci   = load_node(args.ci,   args.model)
    diffs = diff_fields(prod, ci)

    if not diffs:
        print(f"{args.model}: identical (unexpected — would not be state:modified)")
        return

    for field, pv, cv in diffs:
        print(f"\n=== {field} ===")
        lines = difflib.unified_diff(
            pv.splitlines(), cv.splitlines(),
            fromfile=f"prod::{field}", tofile=f"ci::{field}",
            lineterm=""
        )
        print("\n".join(lines))


if __name__ == "__main__":
    main()

# Example output for a legit SQL change
=== raw_code ===
--- prod::raw_code
+++ ci::raw_code
@@ -12,7 +12,7 @@
 SELECT
     o.order_id,
     o.customer_id,
-    o.total_cents          AS gross_cents,
+    o.total_cents          AS gross_cents,
+    o.discount_cents       AS discount_cents,
     o.status
 FROM {{ ref('stg_orders') }} o

=== checksum ===
--- prod::checksum
+++ ci::checksum
@@ -1,4 +1,4 @@
 {
   "name": "sha256",
-  "checksum": "5a2f...9c1d"
+  "checksum": "8d3b...4e77"
 }

Step-by-step explanation.

1. The script loads both manifests, extracts the requested unique_id (dbt's fully-qualified node key), and diffs a curated list of top-level fields. Diffing every field is noisy; the curated list matches what dbt actually fingerprints.
2. raw_code is the post-Jinja-render SQL. A whitespace-only change here still flips the checksum, so if a model is unexpectedly state:modified, this is the first field to inspect. Consider adding an .editorconfig and sqlfmt pre-commit hook to eliminate whitespace-only diffs.
3. checksum is the SHA-256 that dbt actually compares. If checksums match, the model is not state:modified; if they differ, it is. This is the ground-truth field.
4. config catches materialization / cluster_by / tag changes. A model that switches from view to table is state:modified even if the SQL is identical.
5. depends_on.nodes catches upstream ref changes. Adding a new {{ ref('foo') }} marks the model modified; removing one also marks it modified.

Output.

Field	Change type	Marks model as state:modified?
raw_code	any character diff (incl whitespace)	yes
checksum	SHA-256 changed	yes (derived from raw_code + config)
config.materialized	view → table	yes
config.tags	added / removed	yes
depends_on.nodes	new / removed ref	yes
description (yml)	text-only change	no (documentation only)

Rule of thumb. When a PR unexpectedly rebuilds a model that "shouldn't have changed," run the state-diff script against the model — the culprit is almost always a whitespace change in raw_code, a Jinja-var resolution difference, or a macro upgrade that touched the render. Fix the root cause; do not disable slim CI.

Worked example — dbt clone a schema for preview

Detailed explanation. For dashboards, downstream apps, and analysts who want to use the CI schema, an empty pr_<PR>_<branch> schema is not enough — they need it seeded with all the tables that were not rebuilt by slim CI. dbt clone (added in dbt-core 1.6) does this in one command: it zero-copy clones every model in the state manifest that was not rebuilt, so the preview schema is a full mirror of prod overlaid with the rebuilt models. This is the pattern that turns slim CI into a genuine preview environment.

• What dbt clone does. For every model in --state, if the CI schema doesn't already have the model, dbt executes CREATE OR REPLACE <schema>.<model> CLONE prod.<model> (Snowflake) or the equivalent zero-copy on the warehouse.
• Why it's cheap. Snowflake and Databricks Unity Catalog implement zero-copy cloning as a metadata operation; no data is physically copied. A 10 TB table clones in ~1 second and costs $0.
• When to use it. Whenever a human (analyst, PM, stakeholder) needs to SELECT * FROM pr_<PR>_<branch>.orders_mart and expect all the joined tables to be there.

Question. Extend the slim CI workflow to clone unmodified models into the preview schema after the slim build completes.

Input.

Component	Value
Warehouse	Snowflake (or Databricks UC)
Clone command	dbt clone --state prod-manifest/
Target schema	pr__
Runtime for 200-table clone	~5 s (zero-copy metadata op)
Cost for 200-table clone	~$0.00

Code.

# Extension to .github/workflows/dbt-slim.yml — add after "dbt build" step
- name: dbt clone unmodified models into preview schema
  working-directory: dbt
  env:
    DBT_SNOWFLAKE_PRIVATE_KEY: ${{ secrets.SNOWFLAKE_PRIVATE_KEY }}
    DBT_TARGET_SCHEMA: ${{ steps.schema.outputs.schema }}
  run: |
    dbt clone \
      --state ../prod-manifest \
      --target ci \
      --resource-type model \
      --resource-type seed \
      --resource-type snapshot

-- What `dbt clone` executes on Snowflake, per model
CREATE OR REPLACE VIEW ANALYTICS.PR_4712_ADD_ORDERS_V2.STG_CUSTOMERS
  CLONE ANALYTICS.PROD.STG_CUSTOMERS;

CREATE OR REPLACE TABLE ANALYTICS.PR_4712_ADD_ORDERS_V2.DIM_DATE
  CLONE ANALYTICS.PROD.DIM_DATE;

-- ... one CREATE OR REPLACE ... CLONE per model in the state manifest
-- that was not rebuilt by slim CI ...

-- The rebuilt models (from slim CI) are already in the preview schema
-- as freshly-materialized objects, so `CREATE OR REPLACE ... CLONE` on
-- them would overwrite the rebuild. `dbt clone` skips those.

# scripts/verify_preview_schema.py — smoke test after clone
import snowflake.connector

def verify(preview_schema: str) -> None:
    conn = snowflake.connector.connect(
        account="acme-prod",
        user="dbt_ci",
        private_key=load_private_key(),
        role="DBT_CI_ROLE",
        warehouse="DBT_CI_WH_XS",
        database="ANALYTICS",
    )
    with conn.cursor() as cur:
        cur.execute(f"""
            SELECT COUNT(*)
            FROM   ANALYTICS.INFORMATION_SCHEMA.TABLES
            WHERE  TABLE_SCHEMA = %s
        """, (preview_schema.upper(),))
        n_tables = cur.fetchone()[0]

        cur.execute(f"""
            SELECT COUNT(*)
            FROM   ANALYTICS.INFORMATION_SCHEMA.TABLES
            WHERE  TABLE_SCHEMA = 'PROD'
        """)
        n_prod = cur.fetchone()[0]

    assert n_tables == n_prod, (
        f"Preview schema has {n_tables} tables; prod has {n_prod}. "
        "Slim CI + clone should have parity."
    )
    print(f"OK — preview schema {preview_schema} has {n_tables} tables")


if __name__ == "__main__":
    import sys
    verify(sys.argv[1])

Step-by-step explanation.

1. The dbt clone step runs after the slim build so the rebuilt models already exist in the preview schema. Clone only fills in the gaps — the models that were not rebuilt because they were not state:modified.
2. Under the hood, dbt clone emits CREATE OR REPLACE ... CLONE prod.<table> for each unmodified model. Snowflake's zero-copy clone is a metadata-only operation: no data is physically duplicated, no compute is billed, and clone completes in milliseconds per table.
3. --resource-type model --resource-type seed --resource-type snapshot scopes what gets cloned. Skip macros and analyses (they aren't materialised). Include seeds so hard-coded reference tables are present.
4. The verify_preview_schema.py smoke test confirms parity: the number of tables in the preview schema equals the number in prod. If it doesn't, something in the clone step silently failed and the preview would show missing-table errors to analysts.
5. On PR close (see section 5), the cleanup workflow DROP SCHEMA pr_<PR>_<branch> reclaims all of it — the clones are metadata pointers, so dropping the schema releases nothing (no storage was ever allocated), but keeps the catalog tidy.

Output.

Model type	After slim build	After clone
Modified model	freshly materialized	unchanged (skipped by clone)
Unmodified upstream	missing	zero-copy clone from prod
Seed	(only if changed)	zero-copy clone from prod
Snapshot	(only if changed)	zero-copy clone from prod

Rule of thumb. For any dbt preview environment that humans will actually query, follow slim build with dbt clone to backfill unmodified tables. Zero-copy cloning is free on Snowflake and Databricks UC; on BigQuery use table snapshots (also free); on Redshift you must physically CTAS, which costs — reconsider preview UX there or use --defer alone.

Senior interview question on slim CI dbt

A senior interviewer might ask: "You inherit a dbt project with 800 models running full dbt build --target ci on every PR — 55-minute CI, $220 per run, analysts complain they can't review dashboards until merge. Design the slim CI migration: the state artifact contract, the CI workflow, the preview schema, the fallback for macro changes, and the on-call runbook when the prod manifest is missing or stale."

Solution Using slim CI with state comparison, deferral, dbt clone, and a manifest freshness guard

# .github/workflows/dbt-slim-full.yml — production-grade slim CI
name: dbt-slim-ci
on:
  workflow_call:

jobs:
  slim-ci:
    runs-on: ubuntu-latest
    permissions:
      id-token: write
      contents: read
      pull-requests: write
    env:
      DBT_PROFILES_DIR: ./ci
      DBT_PR_NUMBER:    ${{ github.event.pull_request.number }}
      DBT_BRANCH_SLUG:  ${{ github.head_ref }}
      STATE_BUCKET:     dbt-artifacts
      STATE_KEY_PREFIX: prod/latest

    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11' }

      - name: Install dbt
        run: pip install -r dbt/requirements.txt

      - uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: arn:aws:iam::123456789012:role/gha-dbt-ci
          aws-region: us-east-1

      # 1. Download prod manifest — with freshness guard
      - name: Download prod manifest
        id: manifest
        run: |
          mkdir -p prod-manifest
          aws s3 cp s3://${STATE_BUCKET}/${STATE_KEY_PREFIX}/manifest.json prod-manifest/
          aws s3 cp s3://${STATE_BUCKET}/${STATE_KEY_PREFIX}/generated_at.txt prod-manifest/
          age_hours=$(( ( $(date +%s) - $(date -d "$(cat prod-manifest/generated_at.txt)" +%s) ) / 3600 ))
          echo "age_hours=${age_hours}" >> "$GITHUB_OUTPUT"
          if [ "$age_hours" -gt 48 ]; then
            echo "::warning::Prod manifest is ${age_hours}h old; slim CI may over-select"
          fi

      # 2. Detect macro-only or dbt-core changes — fall back to full build
      - name: Detect full-build triggers
        id: mode
        run: |
          if git diff --name-only origin/main | grep -qE '(macros/|dbt_project\.yml|packages\.yml|requirements\.txt)'; then
            echo "mode=full" >> "$GITHUB_OUTPUT"
          else
            echo "mode=slim" >> "$GITHUB_OUTPUT"
          fi

      # 3. Compute preview schema
      - name: Compute preview schema name
        id: schema
        run: |
          slug=$(echo "$DBT_BRANCH_SLUG" | tr '/' '_' | tr -cd '[:alnum:]_' | tr '[:upper:]' '[:lower:]' | cut -c1-40)
          echo "schema=pr_${DBT_PR_NUMBER}_${slug}" >> "$GITHUB_OUTPUT"

      # 4. dbt deps
      - name: dbt deps
        working-directory: dbt
        run: dbt deps

      # 5. dbt build — slim or full
      - name: dbt build
        working-directory: dbt
        env:
          DBT_SNOWFLAKE_PRIVATE_KEY: ${{ secrets.SNOWFLAKE_PRIVATE_KEY }}
          DBT_TARGET_SCHEMA:         ${{ steps.schema.outputs.schema }}
        run: |
          if [ "${{ steps.mode.outputs.mode }}" = "full" ]; then
            echo "Full build (macro / dbt-core change detected)"
            dbt build --target ci --fail-fast
          else
            echo "Slim build (state:modified+)"
            dbt build \
              --select state:modified+ \
              --defer \
              --state ../prod-manifest \
              --target ci \
              --fail-fast
          fi

      # 6. dbt clone — fill preview schema with unmodified prod tables
      - name: dbt clone unmodified models
        if: steps.mode.outputs.mode == 'slim'
        working-directory: dbt
        env:
          DBT_SNOWFLAKE_PRIVATE_KEY: ${{ secrets.SNOWFLAKE_PRIVATE_KEY }}
          DBT_TARGET_SCHEMA:         ${{ steps.schema.outputs.schema }}
        run: |
          dbt clone \
            --state ../prod-manifest \
            --target ci \
            --resource-type model \
            --resource-type seed \
            --resource-type snapshot

      # 7. Upload CI manifest for downstream jobs and post PR comment
      - uses: actions/upload-artifact@v4
        if: always()
        with:
          name: ci-manifest-${{ env.DBT_PR_NUMBER }}
          path: dbt/target/manifest.json

      - name: Post summary to PR
        if: always()
        uses: actions/github-script@v7
        env:
          DBT_MODE:            ${{ steps.mode.outputs.mode }}
          DBT_PREVIEW_SCHEMA:  ${{ steps.schema.outputs.schema }}
          DBT_MANIFEST_AGE_H:  ${{ steps.manifest.outputs.age_hours }}
        with:
          script: |
            const fs = require('fs');
            const j  = JSON.parse(fs.readFileSync('dbt/target/run_results.json', 'utf8'));
            const rows = j.results.slice(0, 40).map(r =>
              `| ${r.unique_id} | ${r.status} | ${r.execution_time.toFixed(2)}s |`
            ).join('\n');
            const body =
              `### dbt CI — PR #${process.env.DBT_PR_NUMBER} (${process.env.DBT_MODE})\n\n` +
              `Preview schema: \`${process.env.DBT_PREVIEW_SCHEMA}\`\n` +
              `Manifest age: ${process.env.DBT_MANIFEST_AGE_H}h\n\n` +
              `| Node | Status | Time |\n|---|---|---|\n${rows}\n`;
            await github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner:  context.repo.owner,
              repo:   context.repo.repo,
              body:   body
            });

# prod dbt job — uploads manifest at end of every successful run
# (runs in Airflow; upload snippet at task boundary)
import boto3
from pathlib import Path
from datetime import datetime, timezone

def upload_prod_manifest():
    s3 = boto3.client("s3")
    bucket = "dbt-artifacts"
    now    = datetime.now(timezone.utc).isoformat()

    s3.upload_file("dbt/target/manifest.json",   bucket, "prod/latest/manifest.json")
    s3.upload_file("dbt/target/run_results.json", bucket, "prod/latest/run_results.json")
    s3.put_object(Bucket=bucket, Key="prod/latest/generated_at.txt", Body=now)

    # Also keep a versioned copy for backfills
    run_id = f"{datetime.now(timezone.utc):%Y%m%dT%H%M%SZ}"
    s3.upload_file("dbt/target/manifest.json", bucket, f"prod/{run_id}/manifest.json")

Step-by-step trace.

Step	Behaviour	Reasoning
Manifest download	pull prod-manifest/manifest.json + generated_at.txt	freshness guard prevents stale state
Manifest age check	warn if >48h old	stale manifest over-selects models
Full-build fallback	trigger on macros/, dbt_project.yml, packages.yml, requirements.txt	slim CI can miss macro impact
Preview schema name	pr__	deterministic; sanitized for Snowflake
dbt build	slim (state:modified+ --defer) or full	mode-aware
dbt clone	zero-copy clone unmodified models into preview	preview UX = full mirror
PR comment	model-by-model results with schema name	reviewers see impact + preview URL
Manifest upload	store dbt/target/manifest.json for downstream	doc-preview / freshness jobs consume it

After deployment, a typical PR rebuilds 4-15 models in ~90 seconds against the preview schema, clones the remaining 785 tables in ~5 seconds (zero-copy metadata operations), and posts a PR comment with the model table + preview schema name; analysts open pr_4712_add_orders_v2.orders_mart in Snowsight and see the new column immediately. Macro-touching PRs fall back to a full build (still under 15 minutes on the XS warehouse with parallel threads). The November invoice drops from ~$9k to ~$200 for CI compute; the average PR is reviewed within 10 minutes of open instead of the next day.

Output:

Metric	Before	After
Median CI time	55 min	2 min
p95 CI cost per PR	$220	$0.35
Preview UX	none	full mirror on PR open
Macro-change safety	none	auto-fallback to full build
Manifest staleness alert	none	warns at 48h
Reviewer feedback loop	next day	~10 min

Why this works — concept by concept:

• state:modified+ --defer --state — the slim CI trinity. state:modified+ scopes the build to only what changed and its descendants; --defer routes unbuilt refs to prod; --state points at the downloaded prod manifest. Together they make the CI job proportionate to the PR, not proportionate to the catalog.
• Prod manifest as artifact contract — the production run uploads manifest.json to a well-known S3 key at the end of every successful run; the CI job downloads it at the top. This is the durable state contract that makes state comparison possible.
• Manifest freshness guard — if the prod manifest is more than 48 hours old, slim CI may over-select (mark models as modified because prod is stale) or under-select (miss real changes because prod-manifest fingerprints don't match current prod). The warning surfaces the problem before it silently distorts the CI output.
• Macro-change full-build fallback — a change to macros/ can affect every model that references the macro without changing the model's SQL text. Slim CI would miss it. Detecting the touched paths and falling back to a full build closes the correctness gap.
• dbt clone for preview parity — slim CI only builds the changed subset; without cloning, the preview schema has 4 tables and analysts see missing-table errors. dbt clone fills the gap in seconds at zero cost via zero-copy metadata.
• Cost — one prod-side upload step (~$0.00), one CI job with slim build + clone (~$0.30 typical), one XS warehouse with AUTO_SUSPEND=60. The eliminated cost is the 55-minute full rebuild ($220 per PR × 30 PRs/day = ~$200k/year). Net O(changed) per PR versus O(catalog) per PR.

SQL
Topic — sql
SQL modeling and warehouse-CI problems

Practice →

ETL Topic — etl ETL problems on incremental modeling

Practice →

---

3. Airflow CI — DAG parse, integration tests, deployable artifacts

DagBag(import_errors={}) is the first gate every Airflow monorepo needs — parse before you promote

The mental model in one line: airflow ci cd is the pattern where every PR that touches airflow/ runs three gates in sequence — a DAG-parse gate that constructs a DagBag in under 30 seconds and asserts zero import_errors, a unit-test matrix that exercises custom operators and hooks with mocked backends, and (optionally) an integration test that runs the changed DAG end-to-end against a LocalExecutor — before the versioned deployable artifact (an OCI image, a dags.zip for MWAA, or an Astronomer deployment bundle) is even built — because a DAG that fails at import time takes down every other DAG in the DagBag, not just its own scheduling. Every senior Airflow deployment has been rescued by the parse gate at least once; skipping it turns "one broken DAG" into "the whole scheduler is red."

The four axes for Airflow CI.

• Correctness. Three levels: parse (does every DAG file import without error?), unit (do custom operators return the right values for mocked inputs?), integration (does a whole DAG execute end-to-end?). Each layer catches a different class of bug. The parse gate is non-negotiable; the unit tests are the coverage layer; the integration test is the "does this DAG actually work?" layer.
• Cost. Airflow CI is essentially free — parse and unit tests run in seconds on ubuntu-latest. The optional integration test is more expensive because it spins a Postgres + LocalExecutor, but still under $0.10 per run. Cost is not the axis that bites for Airflow CI; latency is.
• Latency-to-merge. Parse gate in ~15 seconds is the fastest fail loop of any data-pipeline CI gate. Unit tests in 2-3 minutes. Integration tests in 4-6 minutes. A DAG-only PR should complete CI in under 5 minutes p95.
• Blast radius. A DAG that fails at import time fails every DAG in the DagBag — the scheduler pauses because dagbag_import_error_traceback_depth triggers. The parse gate stops that class of change from reaching the scheduler at all.

The DAG parse gate — 15 seconds, one assertion.

• The mechanism. Instantiate DagBag(dag_folder="airflow/dags", include_examples=False) in a subprocess. If any DAG file raises during import, dagbag.import_errors is a non-empty dict.
• The assertion. assert not dagbag.import_errors, dagbag.import_errors. The assertion message dumps the file paths and tracebacks.
• The trap. DAG files often do work at module import time (query metadata DB, hit Airflow variables, call Variable.get). CI runs without those variables set. Solution: use Variable.get("x", default_var="ci_placeholder") or gate DAG-time work behind if not os.getenv("SKIP_DAG_INIT_WORK").
• The timing. A modern DagBag with 200 DAGs parses in 5-15 seconds. The 30-second Airflow scheduler timeout is your reference point — if CI takes longer, the scheduler will too.

The operator + hook unit-test matrix.

• What it covers. Custom operators (class MyOperator(BaseOperator)), custom hooks (class MyHook(BaseHook)), custom sensors, and utilities. Everything you wrote rather than imported from airflow or a provider package.
• How to structure tests. tests/operators/test_my_operator.py; instantiate the operator with mock upstream context; assert on the return value or the mock's call args.
• What to mock. External services (S3, Snowflake, Kafka) with unittest.mock or moto for AWS. XCom pushes; assert via mock_ti.xcom_push.assert_called_with(...).
• What not to mock. Airflow's own execution machinery — use airflow.models.taskinstance.TaskInstance in "test mode" where possible. Excessive mocking of Airflow itself creates tests that pass but tell you nothing.

The DAG integration test — LocalExecutor + ephemeral Postgres.

• When to run. For DAGs that touch shared resources (a common warehouse table, a critical Kafka topic) or for DAGs that use complex dynamic task mapping. Not every DAG needs one.
• The setup. airflow db init against a Postgres in docker-compose; airflow dags test <dag_id> <execution_date> runs the DAG end-to-end synchronously without a scheduler.
• The runtime. 30 seconds to 6 minutes depending on DAG complexity. Cap the integration-test matrix at the DAGs whose failure would cost more than 6 minutes of CI time to catch.

The deployable artifact — MWAA vs Astronomer vs self-hosted.

• MWAA. dags.zip uploaded to S3; requirements.txt in the same bucket; plugins.zip for custom operators. Versioned by S3 object version + airflow_version field.
• Astronomer. astro deploy builds an OCI image, tags it, and rolls it out to the deployment. Versioned by image digest.
• Self-hosted on Kubernetes. Custom OCI image built from apache/airflow:2.10.x with your DAGs baked in; Helm chart with the image tag.
• The versioning contract. Every deployable artifact carries the git SHA, the branch name, the CI run ID, and the Airflow version. Any deployed artifact must be traceable to a specific commit and CI run.

Common interview probes on Airflow CI.

• "What's the first CI check for Airflow?" — required answer: DAG parse gate (DagBag(import_errors={})).
• "Why does DAG parse fail differently from operator execution?" — required answer: parse errors take down every DAG in the DagBag, not just the broken one.
• "How do you unit-test a custom operator?" — instantiate, provide mock context, assert on return / mock calls.
• "How do you handle Airflow Variables in CI?" — default-value pattern; do not connect to prod metadata DB from CI.
• "How do you version a deployable Airflow artifact?" — git SHA + branch + CI run ID + Airflow version, baked into the image label or zip metadata.

Worked example — GitHub Actions matrix for DAG parse + operator unit tests

Detailed explanation. The canonical Airflow CI matrix: one job for DAG parse (fastest), one job for the operator unit tests (parallel), one job (conditional) for the DAG integration test. Build the whole thing from scratch.

• Matrix. parse (always), unit (always), integration (only if DAG files changed).
• Runners. ubuntu-latest for parse + unit; ubuntu-latest with docker-compose for integration.
• Python. 3.11 (match the Airflow deployment).
• Airflow. 2.10.x pinned to match prod.

Question. Write the GitHub Actions workflow that runs parse + unit tests + optional integration on every airflow-touching PR.

Input.

Job	Runtime	When it runs
dag-parse	15-30 s	every airflow-touching PR
operator-unit	2-3 min	every airflow-touching PR
dag-integration	4-6 min	only if airflow/dags/** changed

Code.

# .github/workflows/airflow.yml — parse, unit, integration
name: airflow-ci
on:
  workflow_call:

jobs:
  # 1. DAG parse gate — fastest fail, 15-30 s
  dag-parse:
    runs-on: ubuntu-latest
    env:
      AIRFLOW_HOME:                       /tmp/airflow
      AIRFLOW__CORE__LOAD_EXAMPLES:       'False'
      AIRFLOW__CORE__UNIT_TEST_MODE:      'True'
      AIRFLOW__CORE__EXECUTOR:            SequentialExecutor
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11' }
      - name: Install Airflow + providers
        run: |
          pip install -r airflow/requirements-ci.txt
      - name: Initialize Airflow metadata DB
        run: airflow db migrate
      - name: Parse every DAG file
        run: python -m airflow.utils.cli_action_loggers ; python scripts/dag_parse_gate.py

  # 2. Operator + hook unit tests — 2-3 min
  operator-unit:
    runs-on: ubuntu-latest
    env:
      AIRFLOW_HOME:                       /tmp/airflow
      AIRFLOW__CORE__LOAD_EXAMPLES:       'False'
      AIRFLOW__CORE__UNIT_TEST_MODE:      'True'
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11' }
      - name: Install
        run: pip install -r airflow/requirements-ci.txt
      - name: Run unit tests
        run: pytest airflow/tests/operators airflow/tests/hooks -v --tb=short

  # 3. DAG integration — only when a DAG file changed
  dag-integration:
    needs: dag-parse
    runs-on: ubuntu-latest
    if: contains(join(github.event.pull_request.changed_files, ','), 'airflow/dags/')
    env:
      AIRFLOW_HOME:                       /tmp/airflow
      AIRFLOW__CORE__LOAD_EXAMPLES:       'False'
      AIRFLOW__CORE__EXECUTOR:            LocalExecutor
      AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql://airflow:airflow@localhost/airflow
    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_USER:     airflow
          POSTGRES_PASSWORD: airflow
          POSTGRES_DB:       airflow
        ports: [ '5432:5432' ]
        options: --health-cmd "pg_isready -U airflow"
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11' }
      - name: Install
        run: pip install -r airflow/requirements-ci.txt
      - name: Init metadata DB
        run: |
          airflow db migrate
          airflow connections add fs_default --conn-type fs
      - name: List changed DAG ids
        id: dags
        run: |
          changed=$(git diff --name-only origin/main -- 'airflow/dags/*.py' | \
                     xargs -I{} python -c 'import ast, sys, pathlib
p = pathlib.Path(sys.argv[1])
tree = ast.parse(p.read_text())
for n in ast.walk(tree):
    if isinstance(n, ast.Assign):
        for t in n.targets:
            if getattr(t, "id", None) == "DAG_ID":
                if isinstance(n.value, ast.Constant):
                    print(n.value.value)' {})
          echo "ids=${changed}" >> "$GITHUB_OUTPUT"
      - name: airflow dags test each changed DAG
        run: |
          for dag_id in ${{ steps.dags.outputs.ids }}; do
            echo "=== testing $dag_id ==="
            airflow dags test "$dag_id" 2026-07-30
          done

# scripts/dag_parse_gate.py — the parse gate assertion
"""Assert every DAG in airflow/dags/ imports without error.

The gate mirrors the check the Airflow scheduler performs — if any DAG
raises during import, the scheduler flags an `import_errors` entry and
the DAG is invisible until fixed. Catching this in CI stops broken DAGs
from ever reaching the scheduler.
"""
import sys
from pathlib import Path
from airflow.models.dagbag import DagBag


def main() -> int:
    dag_folder = Path("airflow/dags").resolve()
    if not dag_folder.exists():
        print(f"ERROR: {dag_folder} does not exist", file=sys.stderr)
        return 2

    dagbag = DagBag(dag_folder=str(dag_folder), include_examples=False)
    if dagbag.import_errors:
        print("DAG parse errors detected:", file=sys.stderr)
        for filepath, err in dagbag.import_errors.items():
            print(f"\n--- {filepath} ---", file=sys.stderr)
            print(err, file=sys.stderr)
        return 1

    print(f"OK — {len(dagbag.dags)} DAGs parsed with 0 import errors")
    for dag_id in sorted(dagbag.dags):
        print(f"  • {dag_id}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

# airflow/tests/operators/test_reverse_etl_operator.py — unit test example
from unittest.mock import MagicMock, patch
import pytest
from airflow.utils.context import Context
from airflow_plugins.operators.reverse_etl import ReverseEtlOperator


@pytest.fixture
def op():
    return ReverseEtlOperator(
        task_id="test_reverse_etl",
        source_table="analytics.customers",
        target_system="hubspot",
        target_object="contacts",
    )


def test_operator_builds_correct_payload(op):
    """Given a source row, the operator should build the expected API payload."""
    row = {"customer_id": 42, "email": "[e-mail]", "first_name": "Ada"}
    payload = op._build_payload(row)
    assert payload == {
        "properties": {
            "email":      "[e-mail]",
            "firstname":  "Ada",
        },
        "id": "42",
    }


@patch("airflow_plugins.operators.reverse_etl.SnowflakeHook")
@patch("airflow_plugins.operators.reverse_etl.HubspotHook")
def test_operator_execute_calls_hubspot(mock_hs, mock_sf, op):
    """Execute should read from Snowflake and POST to Hubspot."""
    mock_sf.return_value.get_records.return_value = [
        {"customer_id": 42, "email": "[e-mail]", "first_name": "Ada"},
    ]
    ctx = MagicMock(spec=Context)
    op.execute(ctx)
    assert mock_hs.return_value.upsert_contact.call_count == 1
    call = mock_hs.return_value.upsert_contact.call_args
    assert call.kwargs["payload"]["properties"]["email"] == "[e-mail]"

Step-by-step explanation.

1. The dag-parse job is the fastest possible fail. It installs Airflow, migrates a SQLite metadata DB (fast), and runs dag_parse_gate.py which instantiates a DagBag and asserts import_errors is empty. Under 30 seconds; catches the highest-impact class of DAG bug.
2. AIRFLOW__CORE__UNIT_TEST_MODE=True tells Airflow to use in-memory connections and skip heavy initialisation; AIRFLOW__CORE__EXECUTOR=SequentialExecutor avoids requiring a real executor. Both are CI-only environment tricks.
3. The operator-unit job runs pytest against airflow/tests/operators and airflow/tests/hooks. Custom operators get mocked hooks (@patch("...SnowflakeHook")); the test asserts on mock_hs.return_value.upsert_contact.call_args. Coverage bar: every custom operator has at least one happy-path and one error-path test.
4. The dag-integration job runs only if a DAG file changed. It spins up Postgres as a GitHub Actions service, migrates the Airflow metadata DB against Postgres, extracts the changed DAG IDs, and runs airflow dags test <dag_id> <date> on each. This runs the DAG end-to-end without the scheduler.
5. The AST-based extraction of DAG_ID from changed DAG files is a pragmatic way to run only the changed DAGs. Alternatives: parse the whole DagBag and match by file path; require every DAG to be tagged with the file it's in. Both work.

Output.

Gate	Runtime	Catches
dag-parse	15-30 s	import errors, syntax errors, missing imports
operator-unit	2-3 min	logic errors in custom operators / hooks
dag-integration	4-6 min	wiring errors, DAG-level task interactions

Rule of thumb. For every Airflow CI setup, run parse first (fastest fail), unit second (broad coverage), integration third (deep coverage on high-value DAGs). Never skip the parse gate; it is 30 seconds and catches the single most common Airflow bug.

Worked example — full-DAG integration test with the LocalExecutor

Detailed explanation. The DAG integration test catches wiring bugs the unit tests cannot: task A returns X but task B expects Y, dynamic task mapping produces the wrong fan-out, or a TaskGroup fails to render under a specific execution date. Walk through a concrete integration test for a DAG that dedupes an S3 landing zone into a Snowflake table.

• DAG under test. s3_to_snowflake_dedupe — 4 tasks: sensor (waits for S3 file) → download → dedupe → merge.
• Fixtures needed. LocalStack S3 (via moto), Snowflake mock (via snowflake.connector mock).
• Execution date. A fixed date so the test is deterministic.

Question. Write an integration test that runs the whole DAG end-to-end and asserts the target Snowflake table receives the expected rows.

Input.

Component	Value
DAG under test	s3_to_snowflake_dedupe
S3 backend	moto (in-process)
Snowflake backend	monkey-patched connector
Execution date	2026-07-30
Fixture rows	3 rows, one duplicate

Code.

# airflow/tests/dags/test_s3_to_snowflake_dedupe.py
import os
import json
from datetime import datetime
from unittest.mock import patch, MagicMock

import pytest
import boto3
from moto import mock_aws

from airflow.models.dagbag import DagBag
from airflow.utils.state import State


@pytest.fixture(scope="module")
def dagbag():
    return DagBag(dag_folder="airflow/dags", include_examples=False)


@pytest.fixture
def s3_landing():
    with mock_aws():
        s3 = boto3.client("s3", region_name="us-east-1")
        s3.create_bucket(Bucket="landing")
        s3.put_object(
            Bucket="landing",
            Key="orders/2026/07/30/orders.jsonl",
            Body=b'{"order_id":1,"total":100}\n'
                 b'{"order_id":2,"total":200}\n'
                 b'{"order_id":1,"total":100}\n',   # duplicate
        )
        yield s3


@pytest.fixture
def snowflake_stub():
    """Replace snowflake.connector.connect with a MagicMock that captures MERGE calls."""
    with patch("snowflake.connector.connect") as mock_connect:
        mock_conn = MagicMock()
        mock_connect.return_value = mock_conn
        mock_cur = MagicMock()
        mock_conn.cursor.return_value.__enter__.return_value = mock_cur
        yield mock_cur


def test_s3_to_snowflake_dedupe_end_to_end(dagbag, s3_landing, snowflake_stub):
    """Run the DAG end-to-end and assert the MERGE receives 2 deduped rows."""
    dag = dagbag.get_dag("s3_to_snowflake_dedupe")
    assert dag is not None, "DAG not found"

    execution_date = datetime(2026, 7, 30)

    # airflow dags test — synchronous, no scheduler needed
    dag.test(execution_date=execution_date)

    # Verify the merge was executed with the deduped payload
    merge_calls = [
        c for c in snowflake_stub.execute.call_args_list
        if "MERGE" in (c.args[0] if c.args else "")
    ]
    assert len(merge_calls) == 1, f"expected 1 MERGE call, got {len(merge_calls)}"

    merge_sql = merge_calls[0].args[0]
    assert "orders_dedupe_stage" in merge_sql
    # Row count sanity
    for ti in dag.get_task_instances():
        assert ti.state in {State.SUCCESS, State.SKIPPED}, (
            f"task {ti.task_id} ended in state {ti.state}"
        )

Step-by-step explanation.

1. The dagbag fixture is module-scoped so parsing runs once for the whole test file. Repeated DagBag() calls per test would inflate CI time by 10-30 seconds per test.
2. moto provides an in-process S3 mock that behaves like real S3 (bucket ops, put_object, list_objects_v2) without any network calls. mock_aws() is the modern moto 5.x context manager.
3. The Snowflake stub replaces snowflake.connector.connect at import time, so any operator code that calls snowflake.connector.connect() receives a MagicMock instead of a real connection. All execute calls are captured in mock_cur.execute.call_args_list.
4. dag.test(execution_date=...) (added in Airflow 2.5) runs the entire DAG synchronously in-process. No scheduler, no executor, no workers — just a straight-through execution. This is the CI-friendly way to run a DAG; airflow dags test from the CLI is the equivalent.
5. The assertions walk the captured Snowflake calls, find the MERGE statement, and verify the deduped row count. Alternative assertion styles: parse the MERGE SQL and check the row count; assert on the task's XCom returns; check the mock's total call count.

Output.

Task	State	Notes
wait_for_s3	SUCCESS	poked once; file present
download_landing	SUCCESS	3 rows from S3
dedupe_orders	SUCCESS	2 unique order_ids
merge_to_snowflake	SUCCESS	MERGE call captured with 2 rows

Rule of thumb. For DAGs that touch high-value shared resources (a warehouse table, a critical topic), invest in a dag.test()-based integration test with moto/monkey-patch stubs. For simpler DAGs, the parse gate + operator unit tests are sufficient. Don't try to integration-test every DAG — the maintenance cost outstrips the value.

Worked example — build and version a deployable Airflow image

Detailed explanation. After CI passes, the last CI job builds the deployable artifact. For self-hosted Kubernetes Airflow, this is an OCI image; for MWAA, it's a dags.zip + requirements.txt uploaded to S3; for Astronomer, it's astro deploy. Walk through the OCI image path — the most versatile.

• Base. apache/airflow:2.10.3-python3.11.
• Overlay. Copy airflow/dags, airflow/plugins, and install airflow/requirements-prod.txt.
• Version labels. git SHA, branch, CI run ID, Airflow version.
• Registry. ghcr.io/acme/airflow tagged with sha-<short> and pr-<PR_NUMBER> (for PRs) or main (for merges).

Question. Write the Dockerfile and the GitHub Actions job that builds, tags, and pushes the versioned image.

Input.

Component	Value
Base image	apache/airflow:2.10.3-python3.11
Registry	ghcr.io/acme/airflow
Tag scheme	sha-<short>, pr-<PR>, main
Labels	git SHA, branch, ci run id, airflow version

Code.

# airflow/Dockerfile
FROM apache/airflow:2.10.3-python3.11

ARG GIT_SHA=unknown
ARG GIT_BRANCH=unknown
ARG CI_RUN_ID=unknown

LABEL org.opencontainers.image.source="https://github.com/acme/data-monorepo"
LABEL org.opencontainers.image.revision="${GIT_SHA}"
LABEL org.opencontainers.image.ref.name="${GIT_BRANCH}"
LABEL ai.pipecode.ci.run.id="${CI_RUN_ID}"
LABEL ai.pipecode.airflow.version="2.10.3"

USER root
RUN apt-get update && apt-get install -y --no-install-recommends git \
 && rm -rf /var/lib/apt/lists/*
USER airflow

COPY --chown=airflow:root requirements-prod.txt /requirements.txt
RUN pip install --no-cache-dir -r /requirements.txt

# DAGs and plugins baked into the image
COPY --chown=airflow:root dags     /opt/airflow/dags
COPY --chown=airflow:root plugins  /opt/airflow/plugins

# Sanity: parse gate at build time
RUN python -c "from airflow.models.dagbag import DagBag; \
               dagbag = DagBag('/opt/airflow/dags', include_examples=False); \
               assert not dagbag.import_errors, dagbag.import_errors; \
               print(f'{len(dagbag.dags)} DAGs, 0 errors')"

# .github/workflows/airflow-build.yml — build + push deployable image
name: airflow-build
on:
  workflow_call:

jobs:
  build:
    runs-on: ubuntu-latest
    permissions:
      contents:  read
      packages:  write
      id-token:  write
    steps:
      - uses: actions/checkout@v4
      - uses: docker/setup-buildx-action@v3
      - uses: docker/login-action@v3
        with:
          registry: ghcr.io
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}

      - name: Compute tags
        id: tags
        run: |
          short_sha=$(git rev-parse --short HEAD)
          echo "sha_tag=ghcr.io/acme/airflow:sha-${short_sha}" >> "$GITHUB_OUTPUT"
          if [ "${{ github.event_name }}" = "pull_request" ]; then
            echo "extra_tag=ghcr.io/acme/airflow:pr-${{ github.event.pull_request.number }}" >> "$GITHUB_OUTPUT"
          else
            echo "extra_tag=ghcr.io/acme/airflow:main" >> "$GITHUB_OUTPUT"
          fi

      - name: Build and push
        uses: docker/build-push-action@v6
        with:
          context: airflow
          push:    true
          tags: |
            ${{ steps.tags.outputs.sha_tag }}
            ${{ steps.tags.outputs.extra_tag }}
          build-args: |
            GIT_SHA=${{ github.sha }}
            GIT_BRANCH=${{ github.ref_name }}
            CI_RUN_ID=${{ github.run_id }}
          cache-from: type=gha
          cache-to:   type=gha,mode=max

Step-by-step explanation.

1. The Dockerfile takes apache/airflow:2.10.3-python3.11 as its base — pinning the patch version prevents accidental upgrades. Use the same tag as prod for tightest parity.
2. Build args flow into LABELs so the running image self-describes: docker inspect ghcr.io/acme/airflow:sha-abc123 | jq '.[0].Config.Labels' reveals git SHA, branch, CI run ID, and Airflow version. Any deployed image can be traced back to a specific commit and CI run.
3. requirements-prod.txt is installed before the DAGs are copied so Docker's layer cache re-uses the pip layer across DAG-only changes — typical CI build drops from 90 s to 15 s on a DAG-only PR.
4. The RUN parse-gate at build time is a belt-and-braces check: even if the CI job's parse gate somehow passed a broken DAG, the image build will fail. This is the last chance to catch an import error before the image goes to the registry.
5. Tags: sha-<short> for every commit (immutable, provenance-preserving) plus pr-<PR_NUMBER> (mutable, points at latest CI build for the PR) or main (mutable, points at latest merge). The immutable sha tag is what deployments actually reference.

Output.

Trigger	Tags pushed	Traceable to
PR open / update	sha-<short>, pr-<PR>	commit SHA + CI run ID
Merge to main	sha-<short>, main	commit SHA + CI run ID
Release tag	sha-<short>, v<VERSION>	commit SHA + git tag

Rule of thumb. For every Airflow deployment, bake DAGs into the image, parse-gate at build time as a safety net, and use immutable sha-<short> tags for actual deployments. The mutable main / pr-<PR> tags are for humans; deployments reference the immutable ones.

Senior interview question on Airflow CI

A senior interviewer might ask: "Your Airflow deployment (MWAA on Airflow 2.10) has 180 DAGs. Once a month, a DAG merges that breaks at import time and the scheduler pauses for the whole DagBag. Design the CI workflow that catches this before merge — the parse gate, the operator unit-test coverage bar, the DAG integration test policy for high-value DAGs, and the deployable-artifact contract including version labels."

Solution Using a parse gate + unit matrix + integration matrix + labelled OCI image

# .github/workflows/airflow-full.yml — production-grade Airflow CI
name: airflow-ci
on:
  workflow_call:

jobs:
  # 1. Parse gate — fastest fail
  parse:
    runs-on: ubuntu-latest
    env:
      AIRFLOW_HOME:                       /tmp/airflow
      AIRFLOW__CORE__LOAD_EXAMPLES:       'False'
      AIRFLOW__CORE__UNIT_TEST_MODE:      'True'
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11' }
      - run: pip install -r airflow/requirements-ci.txt
      - run: airflow db migrate
      - run: python scripts/dag_parse_gate.py

  # 2. Operator/hook unit tests — parallel
  unit:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        suite: [ operators, hooks, sensors, utils ]
    env:
      AIRFLOW_HOME:                       /tmp/airflow
      AIRFLOW__CORE__LOAD_EXAMPLES:       'False'
      AIRFLOW__CORE__UNIT_TEST_MODE:      'True'
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11' }
      - run: pip install -r airflow/requirements-ci.txt
      - name: pytest ${{ matrix.suite }}
        run: pytest airflow/tests/${{ matrix.suite }} -v --tb=short

  # 3. Integration tests for changed DAGs
  integration:
    needs: parse
    runs-on: ubuntu-latest
    if: contains(join(github.event.pull_request.changed_files, ','), 'airflow/dags/')
    services:
      postgres:
        image: postgres:15
        env:
          POSTGRES_USER:     airflow
          POSTGRES_PASSWORD: airflow
          POSTGRES_DB:       airflow
        ports: [ '5432:5432' ]
    env:
      AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql://airflow:airflow@localhost/airflow
      AIRFLOW__CORE__EXECUTOR:             LocalExecutor
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11' }
      - run: pip install -r airflow/requirements-ci.txt
      - run: airflow db migrate
      - run: pytest airflow/tests/integration -v --tb=short

  # 4. Build labelled deployable OCI image
  build-image:
    needs: [ parse, unit, integration ]
    if: always() && needs.parse.result == 'success' && needs.unit.result == 'success' && (needs.integration.result == 'success' || needs.integration.result == 'skipped')
    runs-on: ubuntu-latest
    permissions:
      contents: read
      packages: write
    steps:
      - uses: actions/checkout@v4
      - uses: docker/setup-buildx-action@v3
      - uses: docker/login-action@v3
        with:
          registry: ghcr.io
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}
      - name: Build + push
        uses: docker/build-push-action@v6
        with:
          context: airflow
          push:    true
          tags: |
            ghcr.io/acme/airflow:sha-${{ github.sha }}
            ghcr.io/acme/airflow:pr-${{ github.event.pull_request.number }}
          build-args: |
            GIT_SHA=${{ github.sha }}
            GIT_BRANCH=${{ github.ref_name }}
            CI_RUN_ID=${{ github.run_id }}

# Airflow CI budget contract (per gate)
Gate           p95 latency   Coverage target
─────────────────────────────────────────────
parse          30 s          100% of DAG files
unit-operators 3 min         95% line coverage on custom operators
unit-hooks     2 min         95% line coverage on custom hooks
integration    6 min         100% of DAGs touching shared warehouse tables
build-image    2 min         every merge; every PR

Step-by-step trace.

Gate	Config	Result
parse	DagBag(include_examples=False) + assert not import_errors	catches syntax + import errors
unit (matrix)	pytest airflow/tests/{operators,hooks,sensors,utils}	catches operator logic errors
integration	LocalExecutor + Postgres service + pytest airflow/tests/integration	catches DAG wiring errors
build-image	apache/airflow:2.10.3 + copy DAGs + build-time parse check + labels	produces immutable sha-tagged artifact
Version labels	GIT_SHA / GIT_BRANCH / CI_RUN_ID as OCI LABELs	every deployment traceable

After deployment, the once-a-month "broken DAG paused the scheduler" incident drops to zero — the parse gate catches every broken import before merge. The p95 CI time for an Airflow-only PR is about 5 minutes; the p95 cost is about $0.15. The operator unit-test coverage bar (95%) is enforced by pytest-cov + --cov-fail-under=95. Every deployed image has git SHA + branch + CI run ID + Airflow version in its labels, so any prod incident can be traced back to a specific commit in seconds.

Output:

Metric	Before	After
Broken-DAG incidents	~1/month	~0
p95 CI time (Airflow-only PR)	12 min	5 min
p95 CI cost (Airflow-only PR)	$0.60	$0.15
Operator unit-test coverage	~40%	95% (enforced)
Image provenance	none	git SHA + branch + run ID
DAG integration coverage	0 DAGs	~30 high-value DAGs

Why this works — concept by concept:

• Parse gate first — the single highest-ROI CI check. 30 seconds, one assertion, catches the highest-impact class of Airflow bug (import-time errors that pause the whole DagBag). Every Airflow project should have this on day one.
• Unit-test matrix — pytest matrix parallelises the operator / hook / sensor / util suites. Coverage bar (95%) enforced in CI via --cov-fail-under. Regressions in custom operator logic caught before merge.
• Integration test for shared-resource DAGs — LocalExecutor + Postgres service + dag.test() runs the whole DAG end-to-end in ~1 minute per DAG. Scoped to DAGs that touch shared warehouse tables so the CI time doesn't explode.
• Labelled OCI image — every image carries git SHA, branch, CI run ID, and Airflow version as OCI labels. docker inspect on any deployed image reveals its provenance in one command. Immutable sha-<short> tag is the actual deployment target.
• Build-time parse gate — a belt-and-braces RUN python -c "DagBag(...)" inside the Dockerfile. If the CI parse-gate somehow passed a broken DAG (e.g. Airflow version mismatch), the image build fails before push. Defence in depth.
• Cost — one parse gate ($0.02), one unit matrix ($0.05), one conditional integration ($0.08), one image build ($0.02). Total ~$0.15 per PR. Compared to the "broken DAG paused scheduler for 4 hours" incident cost (~$800 in delayed data), this is essentially free. Net O(changed) per PR versus O(catalog) impact of every prod incident.

Python
Topic — python
Python operator and DAG testing problems

Practice →

API Topic — api-integration API integration problems on Airflow hooks

Practice →

---

4. Spark CI — packaged jobs, unit + integration tests, dry-run submits

spark-submit --deploy-mode client --num-executors 0 is the dry-run gate every Spark CI needs before the real cluster is billed

The mental model in one line: spark ci cd is the pattern where every PR that touches spark/ runs three gates — packaged JAR / wheel that carries the job entry point + its Python / Scala dependencies, local unit tests against a SparkSession.builder.master("local[2]") fixture that catches logic and schema bugs, and a dry-run spark-submit against a small dev cluster (or a docker-compose Spark mini-cluster) that validates cluster reachability, JAR resolution, and config before executors are ever provisioned — because a Spark job that fails after 40 minutes of shuffle costs orders of magnitude more than a job that fails in 30 seconds of local pytest. Every senior Spark deployment has migrated to this pattern once; skipping the local-unit + dry-run gates turns "one bad UDF" into "40-minute cluster time wasted plus a 3 AM page."

The four axes for Spark CI.

• Correctness. Three levels: unit (does the transformation logic produce the right output for known input?), schema (does the job's output schema match downstream expectations?), integration (does spark-submit actually accept the config in the target cluster?). Each layer catches a different class of bug.
• Cost. Local SparkSession is free (runs on the GHA runner). Docker-compose mini-cluster is ~$0.05 per run. Dry-run submit against a dev cluster is ~$0.02. The cost trap is running the real job on the real cluster from CI — that's how a $500 CI run happens.
• Latency-to-merge. Local pytest in 3-4 minutes. Dry-run in 30-60 seconds. A Spark-only PR should complete CI in under 6 minutes p95.
• Blast radius. Local Spark writes to a tempdir. Docker-compose Spark writes to a container-scoped volume. Dry-run submit does not write at all. The real cluster is only touched at deploy time, never at CI time.

The packaged Spark job — one artifact, one entry point.

• PySpark. pip wheel producing job-1.2.3-py3-none-any.whl with entry_points = {"console_scripts": ["run-job = mypkg.entry:main"]}. spark-submit --py-files job-1.2.3.whl --archives venv.tar.gz#env script.py.
• Scala/Java. sbt assembly producing a fat JAR job-1.2.3-assembly.jar. spark-submit --class com.acme.Job job.jar.
• Config. Job-specific YAML (config/prod.yaml, config/dev.yaml) selected by env var; never baked into the artifact.
• Version. Semver tag + git SHA baked into a --version command that prints the SHA. Enables prod-side triage — "which commit is this?" is one flag away.

The local SparkSession fixture — the unit-test workhorse.

• The fixture. pytest session-scoped fixture creating SparkSession.builder.master("local[2]").appName("ci").getOrCreate(). Two cores is enough for logic tests; more is wasted CI time.
• What to test. Pure transformations (def dedupe(df: DataFrame) -> DataFrame:) against known-input DataFrames; UDFs against small Python lists; schema-enforcement (assertSchemaEqual).
• What not to test. Anything requiring shuffle at scale — leave that to the integration test. Anything requiring cluster-specific config (Kryo serializer settings, dynamic allocation) — same.
• Test data. Small in-memory DataFrames (spark.createDataFrame([(1, "a"), (2, "b")], ["id", "name"])) or small JSON fixtures. Never load prod data.

The docker-compose mini-cluster — one CI gate for shuffle behaviour.

• The setup. bitnami/spark:3.5 for master + 2 workers in docker-compose. About 90 seconds to spin up on GHA.
• When to use. Jobs that exercise shuffle, partitioning, or broadcast joins. Local mode does not test these accurately at CI scale.
• When to skip. Pure UDF / column-arithmetic jobs — local mode is enough.

The dry-run spark-submit — the cluster-config gate.

• What it does. Submits the job with --num-executors 0 (or --conf spark.dynamicAllocation.enabled=false --conf spark.executor.instances=0), so the driver starts, resolves the JAR, checks Kerberos / IAM, but no executors spin up.
• What it catches. Missing JAR dependencies, mistyped config keys, cluster-reachability, auth failures, --files resolution errors. Everything that would fail in the first 30 seconds of a real submit.
• What it does not catch. Actual data processing bugs — leave those to the unit tests.
• The Databricks equivalent. POST to /api/2.1/jobs/runs/submit-and-wait with --dry-run=true (or preview via the CLI); validates the job config against the workspace.

Common interview probes on Spark CI.

• "How do you unit-test a PySpark job?" — required answer: pytest fixture with local SparkSession.
• "How do you validate cluster config in CI without spinning up a cluster?" — dry-run submit with --num-executors 0.
• "What's the packaging story for PySpark vs Scala?" — wheel vs fat JAR.
• "How do you test a UDF?" — pytest with the UDF's Python function directly, plus a Spark-side sanity test.
• "How do you avoid running the real job on the real cluster in CI?" — packaging + local mode + dry-run. The real cluster is never touched by CI.

Worked example — pytest against a local SparkSession

Detailed explanation. The canonical PySpark unit test: a session-scoped SparkSession fixture that constructs a local[2] session once per pytest run, plus per-test DataFrames built inline. Walk through a full test module for a dedupe transformation.

• Fixture. spark_session fixture, session scope, local[2].
• Transformation under test. dedupe_orders(df: DataFrame) -> DataFrame.
• Assertions. Row count, schema, specific column values via .collect().

Question. Write the pytest module for a dedupe function with three test cases.

Input.

Test	Input rows	Expected rows
test_basic_dedupe	3 rows (1 duplicate)	2
test_all_duplicate	5 identical rows	1
test_no_duplicates	5 distinct rows	5

Code.

# spark/tests/conftest.py — the shared SparkSession fixture
import pytest
from pyspark.sql import SparkSession


@pytest.fixture(scope="session")
def spark():
    """A local Spark session shared across the pytest session."""
    spark = (
        SparkSession.builder
        .master("local[2]")
        .appName("ci")
        .config("spark.sql.shuffle.partitions", "2")
        .config("spark.ui.enabled", "false")
        .config("spark.sql.session.timeZone", "UTC")
        .getOrCreate()
    )
    yield spark
    spark.stop()

# spark/tests/test_dedupe_orders.py
from pyspark.sql import Row
from mypkg.transforms import dedupe_orders


def test_basic_dedupe(spark):
    df_in = spark.createDataFrame([
        Row(order_id=1, total=100),
        Row(order_id=2, total=200),
        Row(order_id=1, total=100),
    ])
    df_out = dedupe_orders(df_in)
    assert df_out.count() == 2
    ids = sorted(r.order_id for r in df_out.collect())
    assert ids == [1, 2]


def test_all_duplicate(spark):
    df_in = spark.createDataFrame([Row(order_id=42, total=1)] * 5)
    df_out = dedupe_orders(df_in)
    assert df_out.count() == 1


def test_no_duplicates(spark):
    df_in = spark.createDataFrame([
        Row(order_id=i, total=i * 10) for i in range(1, 6)
    ])
    df_out = dedupe_orders(df_in)
    assert df_out.count() == 5


def test_schema_preserved(spark):
    df_in = spark.createDataFrame([
        Row(order_id=1, total=100, extra="x"),
        Row(order_id=1, total=100, extra="x"),
    ])
    df_out = dedupe_orders(df_in)
    assert df_out.schema == df_in.schema

# mypkg/transforms.py — the transformation under test
from pyspark.sql import DataFrame


def dedupe_orders(df: DataFrame) -> DataFrame:
    """Return the DataFrame with duplicate rows removed (all-column key)."""
    return df.dropDuplicates()

Step-by-step explanation.

1. The spark fixture is session-scoped, so the JVM starts up once for the whole test run — not once per test. This is critical: SparkSession.builder.getOrCreate() is ~2-3 seconds of JVM startup, and per-test creation would inflate a 50-test suite by 2-3 minutes.
2. spark.sql.shuffle.partitions=2 overrides the default 200 for the local session; with 2 cores, 200 partitions creates enormous scheduling overhead that dominates the test time. spark.ui.enabled=false skips the Spark UI web server (unneeded in CI).
3. spark.createDataFrame(rows, schema) builds an in-memory DataFrame directly. No file IO, no serialization overhead — the fastest way to construct test data. Row gives typed named-tuple-like construction.
4. .count() and .collect() are the assertion primitives. For larger expected outputs, use .toPandas() + pandas assertions; for schema, use .schema equality or assertSchemaEqual from pyspark.testing.
5. The test_schema_preserved case is a common trap: transformations often silently drop columns or change types. Asserting schema equality catches this regression class.

Output.

Test	Runtime	Fixture cost
First test (JVM startup)	~3 s	one-time
Subsequent tests	~50-200 ms each	none
Full 50-test suite	~30-40 s total	amortised

Rule of thumb. For every PySpark project, ship a session-scoped SparkSession fixture with local[2] + shuffle=2 + ui=false + UTC timezone. Then every unit test is one createDataFrame + one transformation call + one assertion. Under 200 ms per test after the JVM warms.

Worked example — docker-compose Spark mini-cluster

Detailed explanation. For jobs that exercise shuffle, broadcast joins, or specific spark-defaults.conf behaviour, local mode is insufficient — you need a real cluster with executor JVMs. Docker-compose is the CI-friendly way to spin one up in about 90 seconds. Walk through the compose file and a pytest that submits a job against it.

• Compose services. spark-master + 2 × spark-worker + optional spark-history.
• Base image. bitnami/spark:3.5.1 (matches prod Spark version).
• Job submission. docker exec spark-master spark-submit --master spark://spark-master:7077 ….

Question. Write the docker-compose file and a pytest that spins the cluster, submits a job, and asserts the output.

Input.

Component	Value
Spark version	3.5.1
Master + workers	1 + 2
Master port	7077
Job type	JAR
Assertion	output S3 prefix has expected file count

Code.

# spark/tests/integration/docker-compose.yml
services:
  spark-master:
    image: bitnami/spark:3.5.1
    environment:
      - SPARK_MODE=master
      - SPARK_LOCAL_IP=spark-master
    ports:
      - "7077:7077"
      - "8080:8080"
    volumes:
      - ./workdir:/workdir
      - ../../dist:/opt/jobs

  spark-worker-1:
    image: bitnami/spark:3.5.1
    environment:
      - SPARK_MODE=worker
      - SPARK_MASTER_URL=spark://spark-master:7077
      - SPARK_WORKER_MEMORY=2g
      - SPARK_WORKER_CORES=2
    depends_on: [ spark-master ]
    volumes: [ '../../dist:/opt/jobs' ]

  spark-worker-2:
    image: bitnami/spark:3.5.1
    environment:
      - SPARK_MODE=worker
      - SPARK_MASTER_URL=spark://spark-master:7077
      - SPARK_WORKER_MEMORY=2g
      - SPARK_WORKER_CORES=2
    depends_on: [ spark-master ]
    volumes: [ '../../dist:/opt/jobs' ]

# spark/tests/integration/test_dedupe_cluster.py
import subprocess
import time
from pathlib import Path
import pytest


COMPOSE_FILE = Path(__file__).parent / "docker-compose.yml"


@pytest.fixture(scope="module", autouse=True)
def cluster():
    subprocess.check_call(["docker", "compose", "-f", str(COMPOSE_FILE), "up", "-d"])
    # Wait for master to be ready
    for _ in range(30):
        try:
            r = subprocess.run(
                ["docker", "compose", "-f", str(COMPOSE_FILE), "exec", "-T",
                 "spark-master", "curl", "-fsS", "http://localhost:8080"],
                check=True, capture_output=True,
            )
            break
        except subprocess.CalledProcessError:
            time.sleep(2)
    else:
        raise RuntimeError("Spark master did not come up in 60 s")
    yield
    subprocess.check_call(["docker", "compose", "-f", str(COMPOSE_FILE), "down", "-v"])


def test_dedupe_job_end_to_end(cluster, tmp_path):
    """Submit the dedupe job against the mini-cluster and assert output."""
    input_path  = tmp_path / "in"
    output_path = tmp_path / "out"
    input_path.mkdir()
    (input_path / "orders.jsonl").write_text(
        '{"order_id":1,"total":100}\n'
        '{"order_id":2,"total":200}\n'
        '{"order_id":1,"total":100}\n'
    )

    subprocess.check_call([
        "docker", "compose", "-f", str(COMPOSE_FILE), "exec", "-T",
        "spark-master", "spark-submit",
        "--master", "spark://spark-master:7077",
        "--class", "com.acme.DedupeJob",
        "/opt/jobs/dedupe-job-1.0.0.jar",
        "--input",  f"/workdir/in",
        "--output", f"/workdir/out",
    ])

    # Assert output exists and has 2 rows
    result = (tmp_path / "out").glob("*.json")
    files  = list(result)
    assert files, "no output produced"
    lines = sum(1 for f in files for _ in f.read_text().splitlines())
    assert lines == 2, f"expected 2 output rows, got {lines}"

Step-by-step explanation.

1. The compose file defines master + 2 workers using bitnami/spark:3.5.1. Ports 7077 (master) and 8080 (UI) are published so the pytest can healthcheck via HTTP. Volume mounts share the JAR and the working directory between host and containers.
2. The cluster fixture is module-scoped so the cluster spins up once per test module. docker compose up -d runs detached; the polling loop hits the master UI on 8080 until it responds — typically 60-90 seconds.
3. Job submission runs spark-submit inside the master container via docker compose exec. This uses the master's classpath and network — no host-side Spark install needed on the GHA runner.
4. The mounted /workdir shares the input JSON and receives the output — no S3 dependency; everything stays local to the compose stack. This keeps the integration test hermetic and fast.
5. The test asserts on output-file existence and row count. For richer assertions, load the output with pandas or run a validation Spark job. The teardown (down -v) removes volumes so nothing persists.

Output.

Phase	Runtime
docker compose up	45-60 s (first run); 20-30 s (cached image)
Master healthcheck	15-30 s
Job submit + run	30-60 s
docker compose down	5 s
Total (first run)	~2 min

Rule of thumb. Reserve the docker-compose Spark cluster for jobs that need real shuffle / broadcast / partitioning behaviour. For pure-transformation jobs, local mode is enough and finishes in 3-5 minutes total. The compose cluster adds 1-2 minutes for the environment; only pay that cost when the test needs it.

Worked example — dry-run submit that validates cluster config

Detailed explanation. The dry-run submit is the cheapest cluster-level gate: submit the job to a real dev cluster with --num-executors 0, so the driver starts, JAR resolves, config validates, but no executors spin up. If the submit is accepted, the cluster will accept the real submit; if it fails, you catch the config bug in 30 seconds instead of after a 40-minute allocation wait.

• The invocation. spark-submit --master yarn --deploy-mode client --num-executors 0 --class com.acme.Job job.jar --dry-run.
• What passes. JAR resolution, --files resolution, Kerberos ticket, HDFS access.
• What fails. Missing JAR, mistyped --conf, wrong --class, invalid YARN queue.

Question. Write the GHA step and a wrapper script that runs the dry-run submit against a dev EMR cluster.

Input.

Component	Value
Target cluster	dev EMR (via IAM role)
Executors	0 (dry run)
Timeout	60 s
Exit code contract	0 = accepted; non-zero = config bug

Code.

#!/usr/bin/env bash
# scripts/spark_dry_run.sh
set -euo pipefail

JAR="$1"                # path to fat JAR
CLASS="$2"              # main class
CLUSTER_ID="${DEV_EMR_CLUSTER_ID}"

# Submit via EMR CLI with --dry-run-esque args
step_config=$(cat <<EOF
[
  {
    "Type": "Spark",
    "Name": "ci-dry-run-$(git rev-parse --short HEAD)",
    "ActionOnFailure": "CONTINUE",
    "Args": [
      "spark-submit",
      "--deploy-mode", "cluster",
      "--num-executors", "0",
      "--conf", "spark.dynamicAllocation.enabled=false",
      "--conf", "spark.executor.instances=0",
      "--conf", "spark.driver.memory=512m",
      "--conf", "spark.app.name=ci-dry-run",
      "--class", "${CLASS}",
      "${JAR}",
      "--dry-run"
    ]
  }
]
EOF
)

step_id=$(aws emr add-steps \
  --cluster-id "${CLUSTER_ID}" \
  --steps "${step_config}" \
  --query 'StepIds[0]' --output text)

echo "Submitted step ${step_id}; polling…"

for _ in $(seq 1 20); do
  state=$(aws emr describe-step --cluster-id "${CLUSTER_ID}" --step-id "${step_id}" \
          --query 'Step.Status.State' --output text)
  echo "state=${state}"
  case "${state}" in
    COMPLETED)      exit 0 ;;
    FAILED|CANCELLED) exit 1 ;;
    *)              sleep 3 ;;
  esac
done

echo "timeout waiting for dry-run to complete" >&2
exit 2

# GHA step
- name: Spark dry-run against dev cluster
  env:
    AWS_REGION: us-east-1
    DEV_EMR_CLUSTER_ID: ${{ secrets.DEV_EMR_CLUSTER_ID }}
  run: |
    bash scripts/spark_dry_run.sh \
      s3://acme-artifacts/spark/dedupe-job-${{ github.sha }}.jar \
      com.acme.DedupeJob

// spark/src/main/scala/com/acme/DedupeJob.scala — dry-run branch
object DedupeJob {
  def main(args: Array[String]): Unit = {
    val cliArgs = parseArgs(args)
    if (cliArgs.dryRun) {
      println("DRY RUN — validating config only")
      val spark = SparkSession.builder.appName("dedupe-dry-run").getOrCreate()
      // Touch config to force resolution
      println(s"master=${spark.conf.get("spark.master")}")
      println(s"appName=${spark.conf.get("spark.app.name")}")
      spark.stop()
      System.exit(0)
    }
    // real job body …
  }
}

Step-by-step explanation.

1. --num-executors 0 + spark.dynamicAllocation.enabled=false tells the cluster manager "start the driver, but don't allocate any executors." The driver comes up, resolves the JAR, checks config, then exits immediately — 30-60 seconds end to end.
2. Submitting via aws emr add-steps uses the EMR steps API rather than SSHing to the master node. The step ID is returned; the polling loop calls describe-step until the state is terminal.
3. The job's --dry-run flag is recognised in the main class; if set, the job constructs a SparkSession (which forces config resolution) and exits 0. This catches bad --conf values that would otherwise only surface once the real job started reading data.
4. The exit-code contract: 0 = accepted, non-zero = config bug. The GHA step fails fast on non-zero, blocking the PR merge until the config is fixed.
5. The dev cluster is deliberately small (single m5.xlarge master) and shared across CI jobs — it's OK because dry-runs are 30 seconds each. Concurrent CI runs queue up, but no one waits long.

Output.

Failure mode	Caught by dry-run?
Missing JAR at S3	yes (JAR resolution fails)
Mistyped --conf key	partially (some keys validated at driver start)
Wrong --class	yes (ClassNotFoundException at driver init)
Invalid YARN queue	yes (rejected before executors requested)
Data-processing bug	no (need unit tests)
Shuffle-size bug	no (need integration test)

Rule of thumb. For every Spark job, ship a --dry-run code path in the main class that constructs a SparkSession and exits 0. Wire the dry-run into CI as the last gate before merge. This is the cheapest possible "will this run in prod?" check.

Senior interview question on Spark CI

A senior interviewer might ask: "You inherit a Spark job pipeline where CI submits every PR to the prod cluster to test it, burning ~$400/day in wasted compute. Design the CI redesign: packaging, local unit tests, docker-compose integration for high-value jobs, and a dry-run gate. Show me the CI budget, the coverage targets, and the on-call runbook when the dry-run fails on a working PR."

Solution Using packaged JAR + local pytest + docker-compose mini-cluster + dry-run submit

# .github/workflows/spark-full.yml — production-grade Spark CI
name: spark-ci
on:
  workflow_call:

jobs:
  # 1. Package the job — one artifact, reused downstream
  package:
    runs-on: ubuntu-latest
    outputs:
      jar_key: ${{ steps.upload.outputs.key }}
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-java@v4
        with: { distribution: temurin, java-version: '17' }
      - name: sbt assembly
        working-directory: spark
        run: sbt assembly
      - name: Upload JAR
        id: upload
        env:
          KEY: spark/dedupe-job-${{ github.sha }}.jar
        run: |
          aws s3 cp spark/target/scala-2.13/dedupe-job-assembly-*.jar \
                    s3://acme-artifacts/${KEY}
          echo "key=${KEY}" >> "$GITHUB_OUTPUT"

  # 2. Local unit tests — fast, cheap
  unit:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-java@v4
        with: { distribution: temurin, java-version: '17' }
      - uses: actions/setup-python@v5
        with: { python-version: '3.11' }
      - name: Install PySpark for tests
        run: pip install -r spark/requirements-ci.txt
      - name: PySpark unit tests
        run: pytest spark/tests/unit --cov=mypkg --cov-fail-under=90
      - name: Scala unit tests
        working-directory: spark
        run: sbt test

  # 3. Integration — only for shuffle-heavy jobs
  integration:
    if: contains(join(github.event.pull_request.changed_files, ','), 'spark/jobs/')
    runs-on: ubuntu-latest
    needs: package
    steps:
      - uses: actions/checkout@v4
      - name: docker-compose spark cluster + integration test
        run: |
          cd spark/tests/integration
          docker compose up -d
          for i in $(seq 1 30); do
            docker compose exec -T spark-master curl -fsS http://localhost:8080 && break
            sleep 2
          done
          pip install pytest pyspark
          pytest .
          docker compose down -v

  # 4. Dry-run submit — cluster gate
  dry-run:
    if: needs.unit.result == 'success' && needs.package.result == 'success'
    needs: [ package, unit ]
    runs-on: ubuntu-latest
    permissions:
      id-token: write
      contents: read
    steps:
      - uses: actions/checkout@v4
      - uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: arn:aws:iam::123456789012:role/gha-spark-dry-run
          aws-region: us-east-1
      - env:
          DEV_EMR_CLUSTER_ID: ${{ secrets.DEV_EMR_CLUSTER_ID }}
        run: |
          bash scripts/spark_dry_run.sh \
            s3://acme-artifacts/${{ needs.package.outputs.jar_key }} \
            com.acme.DedupeJob

# Spark CI budget contract
Gate           p95 latency   Cost budget   Coverage
─────────────────────────────────────────────────────
package         60 s         $0.02         100% jobs
unit            4 min        $0.03         90% line coverage
integration     6 min        $0.05         high-value jobs only
dry-run         60 s         $0.02         100% jobs before merge
Total (typical) 6 min        $0.12

Step-by-step trace.

Gate	Config	Result
package	sbt assembly + upload to s3	one immutable JAR per commit SHA
unit	pytest + sbt test, 90% coverage floor	catches logic + schema bugs
integration	docker-compose spark-3.5.1 mini-cluster	catches shuffle + join bugs
dry-run	EMR add-steps with --num-executors 0	catches cluster-config bugs
Prod cluster touched?	never in CI	cost isolated to deploy

After deployment, CI submits zero real jobs to the prod cluster — the daily $400 wasted-compute bill goes to zero. The p95 CI time for a Spark-only PR is about 6 minutes; the p95 cost is about $0.12. Unit test coverage on the transformation package is enforced at 90% via --cov-fail-under=90. Every merged commit has a JAR at s3://acme-artifacts/spark/dedupe-job-<sha>.jar, so prod rollouts and rollbacks are one S3 pointer change.

Output:

Metric	Before	After
Wasted-compute bill (CI)	~$400/day	$0
p95 CI time (Spark-only PR)	42 min	6 min
p95 CI cost per PR	$8	$0.12
Unit-test coverage	~30%	90% (enforced)
Job-versioning story	none	immutable JAR per SHA
Broken-config incidents	~2/week	~0

Why this works — concept by concept:

• Packaged JAR / wheel per commit — one immutable artifact per SHA, keyed by git SHA. Every downstream test uses the same artifact; prod deploys reference it by SHA. This kills the "which version is in prod?" question at its root.
• Local SparkSession fixture — fastest possible unit-test loop. local[2] + shuffle=2 + ui=false keeps the fixture 3 seconds to warm and 100-200 ms per test after. The 90% coverage floor is a contract, not a hope; --cov-fail-under=90 enforces it.
• Docker-compose mini-cluster — scoped to jobs that actually need real shuffle. Adds ~2 minutes to CI when it runs, but stays gated on spark/jobs/** path filter so most PRs skip it.
• Dry-run submit — the cheapest cluster-level gate. --num-executors 0 starts the driver, resolves everything, exits — 30-60 seconds. Catches config, JAR, class, queue bugs before the real submit would.
• Never touch the prod cluster from CI — the invariant. All gates run on the GHA runner, a local docker-compose stack, or a small dev EMR cluster. Prod compute is only billed for prod work.
• Cost — $0.02 package + $0.03 unit + optional $0.05 integration + $0.02 dry-run = $0.12 per PR typical. Compared to the $8/PR baseline (mostly wasted prod-cluster time), that's a 60× cost reduction. Net O(local) per PR versus O(cluster) per PR.

Python
Topic — python
Python data-transformation testing problems

Practice →

Streaming Topic — streaming Streaming and Spark job problems

Practice →

---

5. Preview environments — one branch, one warehouse schema

pr_<PR>_<branch> schema on Snowflake / Databricks UC — created on PR open, dropped on PR close

The mental model in one line: dbt preview environment is the pattern where every open pull request maps to its own throwaway warehouse schema named pr_<PR>_<branch> — created on PR open via CREATE SCHEMA ... CLONE prod (Snowflake / Databricks UC zero-copy), targeted by the slim-CI dbt build, exposed to analysts as a preview URL, and auto-dropped by a cleanup workflow when the PR closes — so every PR has an isolated, near-prod dataset for review without ever touching the prod schema and without paying for full-copy storage.

The four axes for preview environments.

• Correctness. The preview schema is a near-prod dataset — clones of every unmodified table, freshly-built copies of changed models. Analysts querying the preview see exactly what production will look like after merge.
• Cost. Snowflake and Databricks UC zero-copy clones are metadata operations — no data is physically duplicated, no storage is billed. XS warehouse with AUTO_SUSPEND=60 keeps compute cost under $0.10 per preview session.
• Latency-to-preview. Slim build (~90 s) + clone (~5 s) + link posted to PR (~2 s) = analyst has the preview URL within 100 seconds of PR open.
• Blast radius. Each preview lives in its own schema; concurrent PRs never collide. Cleanup on PR close reclaims the catalog entry.

Schema naming — deterministic and cleanable.

• Format. pr_<PR_NUMBER>_<BRANCH_SLUG> — e.g. pr_4712_add_orders_v2.
• Branch slug rules. Lowercase, strip non-alphanumeric, cap at 40 characters. Snowflake identifier limit is 255; capping at 40 leaves room for the pr_<num>_ prefix.
• Case. Lowercase everywhere — Snowflake folds unquoted identifiers to uppercase; dbt writes lowercase by default. Consistency across CI + prod avoids "table not found" surprises.
• Cleanup. A cleanup workflow triggers on pull_request closed (both merge and reject) and runs DROP SCHEMA IF EXISTS pr_<PR>_<branch> CASCADE.

Snowflake zero-copy CLONE.

• Command. CREATE OR REPLACE SCHEMA analytics.pr_4712_add_orders_v2 CLONE analytics.prod;.
• Semantics. All tables, views, and pipes in the source schema are cloned as metadata pointers. Reads work immediately; writes copy-on-write.
• Cost. $0 storage until a write happens. A read-only preview costs storage $0.

Databricks Unity Catalog SCHEMA CLONE.

• Command. CREATE SCHEMA <catalog>.pr_4712_add_orders_v2 DEEP CLONE <catalog>.prod; for deep, or SHALLOW CLONE for zero-copy.
• Semantics. Shallow clone is metadata-only (like Snowflake); deep clone physically copies. Prefer shallow for previews.
• Cost. Shallow clone = $0. Deep clone = full storage cost.

Ephemeral compute — AUTO_SUSPEND=60 is the invariant.

• Snowflake. WAREHOUSE_SIZE=XSMALL, AUTO_SUSPEND=60, AUTO_RESUME=TRUE. XS warehouse is $1/hour; suspending after 60 s of idle keeps a typical CI session under 5 minutes of billed compute.
• Databricks. Interactive clusters with autotermination_minutes: 10; SQL warehouses with auto_stop_mins: 5.
• BigQuery. On-demand billing means no explicit suspend — every query is priced independently.

Cleanup on PR close — the non-negotiable half of the pattern.

• The trigger. on: pull_request: types: [closed] in GitHub Actions.
• The action. DROP SCHEMA IF EXISTS pr_<PR>_<branch> CASCADE — idempotent.
• The safety net. Nightly cron that drops any pr_% schema older than 14 days — catches PRs abandoned without close.

Common interview probes on preview environments.

• "How do you avoid full-copy storage cost for previews?" — zero-copy clone (Snowflake CREATE SCHEMA CLONE / Databricks SHALLOW CLONE).
• "How do you name preview schemas?" — pr_<PR>_<branch_slug>, deterministic, cleanable.
• "What if a PR is abandoned?" — cleanup workflow on closed + nightly cron for pr_% older than 14 days.
• "How do analysts access the preview?" — preview URL posted to PR comment; Snowsight / Databricks SQL link with the schema pre-selected.
• "What's the cost per preview?" — $0 storage (zero-copy), $0.05-$0.30 compute for a typical review session.

Worked example — create-preview-schema GitHub Action

Detailed explanation. The create-preview workflow runs on PR open (and on every push to the PR) and produces the pr_<PR>_<branch> schema. It's idempotent — CREATE OR REPLACE — so a re-push refreshes the preview against the latest changes.

• Trigger. pull_request types: [opened, synchronize, reopened].
• Action. Snowflake CREATE OR REPLACE SCHEMA ... CLONE prod.
• Grant. Read-only access to the schema for the analyst role.

Question. Write the workflow and the SQL that creates a preview schema for a PR.

Input.

Component	Value
Trigger	pull_request opened / synchronize / reopened
Warehouse	DBT_CI_WH_XS (AUTO_SUSPEND=60)
Source schema	ANALYTICS.PROD
Target schema	ANALYTICS.pr__
Analyst grant	SELECT to ANALYST_ROLE

Code.

# .github/workflows/preview-schema-create.yml
name: preview-schema-create
on:
  workflow_call:

jobs:
  create-preview:
    runs-on: ubuntu-latest
    permissions:
      id-token: write
      contents: read
      pull-requests: write
    env:
      PR_NUMBER:    ${{ github.event.pull_request.number }}
      BRANCH_SLUG:  ${{ github.head_ref }}
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11' }
      - run: pip install snowflake-connector-python
      - name: Compute schema name
        id: schema
        run: |
          slug=$(echo "$BRANCH_SLUG" | tr '/' '_' | tr -cd '[:alnum:]_' | tr '[:upper:]' '[:lower:]' | cut -c1-40)
          echo "name=pr_${PR_NUMBER}_${slug}" >> "$GITHUB_OUTPUT"
      - name: Create preview schema
        env:
          SNOWFLAKE_ACCOUNT:     ${{ secrets.SNOWFLAKE_ACCOUNT }}
          SNOWFLAKE_USER:        dbt_ci
          SNOWFLAKE_PRIVATE_KEY: ${{ secrets.SNOWFLAKE_PRIVATE_KEY }}
          SCHEMA_NAME:           ${{ steps.schema.outputs.name }}
        run: python scripts/create_preview_schema.py
      - name: Post preview URL to PR
        uses: actions/github-script@v7
        env:
          SCHEMA_NAME: ${{ steps.schema.outputs.name }}
        with:
          script: |
            const url = `https://acme.snowflakecomputing.com/console#/data/databases/ANALYTICS/schemas/${process.env.SCHEMA_NAME.toUpperCase()}`;
            const body = `Preview schema created: \`ANALYTICS.${process.env.SCHEMA_NAME}\`\n\n[Open in Snowsight](${url})`;
            await github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo:  context.repo.repo,
              body:  body,
            });

# scripts/create_preview_schema.py
import os
import snowflake.connector

def main() -> None:
    schema = os.environ["SCHEMA_NAME"].upper()

    conn = snowflake.connector.connect(
        account=os.environ["SNOWFLAKE_ACCOUNT"],
        user=os.environ["SNOWFLAKE_USER"],
        private_key=_load_pk(os.environ["SNOWFLAKE_PRIVATE_KEY"]),
        role="DBT_CI_ROLE",
        warehouse="DBT_CI_WH_XS",
        database="ANALYTICS",
    )
    with conn.cursor() as cur:
        cur.execute(f"CREATE OR REPLACE SCHEMA ANALYTICS.{schema} CLONE ANALYTICS.PROD")
        cur.execute(f"GRANT USAGE  ON SCHEMA ANALYTICS.{schema} TO ROLE ANALYST_ROLE")
        cur.execute(f"GRANT SELECT ON ALL TABLES IN SCHEMA ANALYTICS.{schema} TO ROLE ANALYST_ROLE")
        cur.execute(f"GRANT SELECT ON ALL VIEWS  IN SCHEMA ANALYTICS.{schema} TO ROLE ANALYST_ROLE")
    print(f"OK — created {schema}")


def _load_pk(pem: str):
    from cryptography.hazmat.primitives.serialization import load_pem_private_key
    return load_pem_private_key(pem.encode(), password=None).private_bytes(
        encoding=__import__("cryptography").hazmat.primitives.serialization.Encoding.DER,
        format=__import__("cryptography").hazmat.primitives.serialization.PrivateFormat.PKCS8,
        encryption_algorithm=__import__("cryptography").hazmat.primitives.serialization.NoEncryption(),
    )


if __name__ == "__main__":
    main()

Step-by-step explanation.

1. The workflow triggers on PR open + every synchronize (push). CREATE OR REPLACE is idempotent — re-running refreshes the clone against the current prod state.
2. Schema name construction sanitises the branch (lowercase, alphanumeric + underscore only, capped at 40 chars). Snowflake identifiers are case-folded to uppercase unless quoted; sticking to lowercase in code + uppercase at the SQL layer keeps everything predictable.
3. CREATE OR REPLACE SCHEMA ... CLONE ANALYTICS.PROD is the zero-copy clone. All tables, views, sequences, pipes in PROD become pointers in the new schema. No data copied; no storage billed until write.
4. The GRANT statements give the analyst role read access to the preview schema — Snowsight browses will work, dashboards can be repointed, and stakeholders can SELECT. No write access, so an analyst cannot accidentally mutate the preview.
5. The PR comment includes a Snowsight deep-link with the schema pre-selected, so reviewers click through to the preview in one hop.

Output.

Step	Runtime	Cost
Snowflake connect	~2 s	$0
CREATE OR REPLACE SCHEMA ... CLONE	~3 s	$0 (metadata op)
GRANT statements	~1 s	$0
Post PR comment	~1 s	$0
Total	~10 s	$0

Rule of thumb. Create the preview schema via zero-copy clone on PR open; make it idempotent (CREATE OR REPLACE); post the schema name and a Snowsight link to the PR. Analysts get the preview URL as fast as CI can post a comment.

Worked example — destroy-preview-schema on PR close

Detailed explanation. The cleanup workflow is the non-negotiable other half of the pattern. Without it, preview schemas accumulate indefinitely and eventually clutter the catalog. With it, catalog footprint is bounded to the count of open PRs.

• Trigger. pull_request type: closed (fires on both merge and reject).
• Action. DROP SCHEMA IF EXISTS ... CASCADE — idempotent; safe even if the schema was already dropped.

Question. Write the cleanup workflow.

Input.

Component	Value
Trigger	pull_request closed
Action	DROP SCHEMA IF EXISTS pr__ CASCADE
Cleanup guarantee	fires whether PR merged or rejected

Code.

# .github/workflows/preview-schema-drop.yml
name: preview-schema-drop
on:
  pull_request:
    types: [ closed ]

jobs:
  drop-preview:
    runs-on: ubuntu-latest
    permissions:
      id-token: write
      contents: read
    env:
      PR_NUMBER:   ${{ github.event.pull_request.number }}
      BRANCH_SLUG: ${{ github.head_ref }}
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11' }
      - run: pip install snowflake-connector-python
      - name: Compute schema name
        id: schema
        run: |
          slug=$(echo "$BRANCH_SLUG" | tr '/' '_' | tr -cd '[:alnum:]_' | tr '[:upper:]' '[:lower:]' | cut -c1-40)
          echo "name=pr_${PR_NUMBER}_${slug}" >> "$GITHUB_OUTPUT"
      - name: Drop preview schema
        env:
          SNOWFLAKE_ACCOUNT:     ${{ secrets.SNOWFLAKE_ACCOUNT }}
          SNOWFLAKE_USER:        dbt_ci
          SNOWFLAKE_PRIVATE_KEY: ${{ secrets.SNOWFLAKE_PRIVATE_KEY }}
          SCHEMA_NAME:           ${{ steps.schema.outputs.name }}
        run: python scripts/drop_preview_schema.py

# scripts/drop_preview_schema.py
import os
import snowflake.connector

def main() -> None:
    schema = os.environ["SCHEMA_NAME"].upper()
    conn = snowflake.connector.connect(
        account=os.environ["SNOWFLAKE_ACCOUNT"],
        user=os.environ["SNOWFLAKE_USER"],
        private_key=_load_pk(os.environ["SNOWFLAKE_PRIVATE_KEY"]),
        role="DBT_CI_ROLE",
        warehouse="DBT_CI_WH_XS",
        database="ANALYTICS",
    )
    with conn.cursor() as cur:
        cur.execute(f"DROP SCHEMA IF EXISTS ANALYTICS.{schema} CASCADE")
    print(f"OK — dropped {schema} (or it did not exist)")


def _load_pk(pem: str):
    from cryptography.hazmat.primitives.serialization import (
        load_pem_private_key, Encoding, PrivateFormat, NoEncryption,
    )
    return load_pem_private_key(pem.encode(), password=None).private_bytes(
        encoding=Encoding.DER, format=PrivateFormat.PKCS8,
        encryption_algorithm=NoEncryption(),
    )


if __name__ == "__main__":
    main()

-- Nightly safety-net cron — drop pr_% schemas older than 14 days
DECLARE
  schemas_to_drop VARRAY(2000);
BEGIN
  SELECT ARRAY_AGG(SCHEMA_NAME) INTO :schemas_to_drop
  FROM   ANALYTICS.INFORMATION_SCHEMA.SCHEMATA
  WHERE  SCHEMA_NAME LIKE 'PR_%'
    AND  CREATED < DATEADD(day, -14, CURRENT_TIMESTAMP());

  FOR i IN 0 TO ARRAY_SIZE(:schemas_to_drop) - 1 DO
    EXECUTE IMMEDIATE 'DROP SCHEMA IF EXISTS ANALYTICS.' || GET(:schemas_to_drop, i) || ' CASCADE';
  END FOR;
END;

Step-by-step explanation.

1. The pull_request: closed trigger fires whether the PR is merged or rejected. GitHub does not distinguish between merged and abandoned closes in the trigger itself — both should trigger cleanup because both leave the preview schema orphaned.
2. DROP SCHEMA IF EXISTS ... CASCADE is idempotent: IF EXISTS prevents an error if the schema was already dropped (concurrent runs, manual cleanup, etc.). CASCADE drops all objects in the schema; for zero-copy clones there's no storage impact.
3. The Snowflake schema name is uppercased at the SQL layer because Snowflake case-folds unquoted identifiers. Consistency: lowercase in Python + workflow logic, uppercase at SQL.
4. The nightly safety-net cron catches schemas that escaped the pull_request trigger — e.g. PRs closed while GitHub Actions was down, PRs deleted rather than closed, PRs from forks that bypass the trigger. Bounded staleness = 14 days.
5. The DROP is essentially free — the source schema (PROD) is untouched; only the metadata pointer schema is removed. Runtime is under 5 seconds even for schemas with thousands of tables.

Output.

Event	Cleanup path
PR merged	pull_request-closed trigger → DROP within 30 s
PR rejected	pull_request-closed trigger → DROP within 30 s
PR abandoned (never closed)	nightly cron catches at day 14
Fork PR (limited trigger)	nightly cron catches at day 14
GitHub Actions outage	nightly cron catches at day 14

Rule of thumb. Ship the cleanup workflow the same day as the create workflow. Add the nightly cron the same week. A preview-environment system without cleanup accumulates schemas until someone screams; a system with both cleanup layers stays bounded forever.

Worked example — seeding a preview from a sampled production dataset

Detailed explanation. For sensitive workloads (PII, revenue data) where the preview should not contain full prod data, seed the preview from a sampled subset instead of a zero-copy clone. This trades preview fidelity for privacy.

• The sampling. CREATE TABLE ... AS SELECT * FROM prod.orders SAMPLE (5) — Snowflake's 5% random sample.
• The PII handling. Redact or hash sensitive columns during the sample.

Question. Write the sampling script that seeds a preview schema with 5% of prod data, redacting emails.

Input.

Component	Value
Source	ANALYTICS.PROD
Sample rate	5%
Redacted columns	orders.customer_email → sha256
Target	ANALYTICS.pr__

Code.

# scripts/seed_sampled_preview.py
import os
import snowflake.connector

TABLES_TO_SAMPLE = {
    "ORDERS":     ["customer_email"],  # columns to redact
    "CUSTOMERS":  ["email", "phone"],
    "SHIPMENTS":  [],
}


def main() -> None:
    schema = os.environ["SCHEMA_NAME"].upper()
    conn = snowflake.connector.connect(
        account=os.environ["SNOWFLAKE_ACCOUNT"],
        user=os.environ["SNOWFLAKE_USER"],
        private_key=_load_pk(os.environ["SNOWFLAKE_PRIVATE_KEY"]),
        role="DBT_CI_ROLE",
        warehouse="DBT_CI_WH_XS",
        database="ANALYTICS",
    )
    with conn.cursor() as cur:
        cur.execute(f"CREATE SCHEMA IF NOT EXISTS ANALYTICS.{schema}")
        for table, redact_cols in TABLES_TO_SAMPLE.items():
            col_expr = _column_expressions(cur, table, redact_cols)
            sql = (
                f"CREATE OR REPLACE TABLE ANALYTICS.{schema}.{table} AS "
                f"SELECT {col_expr} FROM ANALYTICS.PROD.{table} SAMPLE (5)"
            )
            cur.execute(sql)
            print(f"OK — seeded {schema}.{table}")


def _column_expressions(cur, table: str, redact_cols: list[str]) -> str:
    cur.execute(f"""
        SELECT COLUMN_NAME FROM ANALYTICS.INFORMATION_SCHEMA.COLUMNS
        WHERE  TABLE_SCHEMA = 'PROD' AND TABLE_NAME = '{table}'
        ORDER  BY ORDINAL_POSITION
    """)
    cols = [r[0] for r in cur.fetchall()]
    parts = []
    for c in cols:
        if c.lower() in {r.lower() for r in redact_cols}:
            parts.append(f"SHA2({c}, 256) AS {c}")
        else:
            parts.append(c)
    return ", ".join(parts)


def _load_pk(pem: str):
    from cryptography.hazmat.primitives.serialization import (
        load_pem_private_key, Encoding, PrivateFormat, NoEncryption,
    )
    return load_pem_private_key(pem.encode(), password=None).private_bytes(
        encoding=Encoding.DER, format=PrivateFormat.PKCS8,
        encryption_algorithm=NoEncryption(),
    )


if __name__ == "__main__":
    main()

Step-by-step explanation.

1. The sampling script iterates a table → redact-cols dict, so adding a new table is a one-line change. Column names are looked up dynamically from INFORMATION_SCHEMA.COLUMNS to avoid schema drift.
2. SAMPLE (5) is Snowflake's Bernoulli row sample: each row has a 5% independent chance of inclusion. For a 100M-row table this yields ~5M rows; runtime is roughly proportional to the sample size, not the source size.
3. Redacted columns are SHA-256 hashed rather than dropped, so downstream JOIN semantics survive. SHA2(email, 256) keeps the column at its original width and preserves uniqueness for de-dupe logic.
4. CREATE OR REPLACE TABLE (not zero-copy clone) is used because the redaction requires actual data materialisation. Storage cost applies — a 5% sample of a 1 TB table costs ~50 GB. Retention is the same 14-day nightly cron.
5. This pattern is the exception, not the rule — most previews use zero-copy clones. Reserve the sampled-preview path for PII-sensitive schemas, and document which tables are redacted.

Output.

Table	Rows in prod	Rows in preview	Redacted cols
ORDERS	100M	~5M	customer_email → sha256
CUSTOMERS	10M	~500K	email, phone → sha256
SHIPMENTS	50M	~2.5M	(none)

Rule of thumb. Default to zero-copy clone; use sampled + redacted only for PII-sensitive schemas. Document the redaction list in a versioned config. Analysts querying the preview must know which columns are hashes.

Senior interview question on preview environments

A senior interviewer might ask: "Your dbt project has 800 models and 30 open PRs at any time. Design the preview-environment system: schema naming, creation, cleanup, per-PR cost bound, and the plan for PII-sensitive schemas. Include the runbook when the cleanup workflow fails."

Solution Using create + drop workflows + nightly safety-net + optional sampled seed

# .github/workflows/preview-envs.yml — full preview lifecycle
name: preview-envs
on:
  pull_request:
    types: [ opened, synchronize, reopened, closed ]
  schedule:
    - cron: '0 6 * * *'   # nightly safety-net at 06:00 UTC

jobs:
  create:
    if: github.event.action != 'closed' && github.event_name == 'pull_request'
    uses: ./.github/workflows/preview-schema-create.yml

  drop:
    if: github.event.action == 'closed'
    uses: ./.github/workflows/preview-schema-drop.yml

  nightly-safety-net:
    if: github.event_name == 'schedule'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11' }
      - run: pip install snowflake-connector-python
      - name: Drop pr_% schemas older than 14 days
        env:
          SNOWFLAKE_ACCOUNT:     ${{ secrets.SNOWFLAKE_ACCOUNT }}
          SNOWFLAKE_USER:        dbt_ci
          SNOWFLAKE_PRIVATE_KEY: ${{ secrets.SNOWFLAKE_PRIVATE_KEY }}
        run: python scripts/preview_safety_net.py

# scripts/preview_safety_net.py — bounds preview-schema staleness
import os
import snowflake.connector

def main() -> None:
    conn = snowflake.connector.connect(
        account=os.environ["SNOWFLAKE_ACCOUNT"],
        user=os.environ["SNOWFLAKE_USER"],
        private_key=_load_pk(os.environ["SNOWFLAKE_PRIVATE_KEY"]),
        role="DBT_CI_ROLE",
        warehouse="DBT_CI_WH_XS",
        database="ANALYTICS",
    )
    with conn.cursor() as cur:
        cur.execute("""
            SELECT SCHEMA_NAME
            FROM   ANALYTICS.INFORMATION_SCHEMA.SCHEMATA
            WHERE  SCHEMA_NAME LIKE 'PR_%'
              AND  CREATED < DATEADD(day, -14, CURRENT_TIMESTAMP())
        """)
        stale = [r[0] for r in cur.fetchall()]
        for s in stale:
            cur.execute(f"DROP SCHEMA IF EXISTS ANALYTICS.{s} CASCADE")
            print(f"safety-net dropped {s}")
    print(f"OK — {len(stale)} schemas cleaned")


def _load_pk(pem: str):
    from cryptography.hazmat.primitives.serialization import (
        load_pem_private_key, Encoding, PrivateFormat, NoEncryption,
    )
    return load_pem_private_key(pem.encode(), password=None).private_bytes(
        encoding=Encoding.DER, format=PrivateFormat.PKCS8,
        encryption_algorithm=NoEncryption(),
    )


if __name__ == "__main__":
    main()

Step-by-step trace.

Event	Workflow	Result
PR open / sync	preview-schema-create	CREATE OR REPLACE SCHEMA … CLONE PROD; post URL
PR merged	preview-schema-drop	DROP SCHEMA IF EXISTS … CASCADE
PR rejected	preview-schema-drop	same
PR abandoned	nightly cron at day 14	DROP SCHEMA IF EXISTS … CASCADE
GitHub Actions outage	nightly cron	catches later
Analyst review	Snowsight URL from PR comment	reads preview schema
PII schema	scripts/seed_sampled_preview.py	5% sample + SHA-256 redaction

After deployment, the catalog holds exactly one schema per open PR plus prod; the November catalog audit shows zero orphaned pr_% schemas older than 14 days. Median preview creation time is 10 seconds; median compute cost per PR is $0.05 (analyst review session on the XS warehouse). Analysts open the preview URL in one click from the PR description.

Output:

Metric	Value
Preview creation latency	~10 s
Preview compute cost per PR	~$0.05
Preview storage cost	$0 (zero-copy)
Orphaned schemas (14-day window)	0
Cleanup coverage	100% (trigger + cron)
PII redaction	opt-in per schema

Why this works — concept by concept:

• Deterministic schema naming — pr_<PR>_<branch_slug> is generated from GitHub context. Any script that knows the PR number can construct the name. This makes create + drop trivially symmetric.
• Zero-copy CLONE — Snowflake and Databricks UC implement clone as metadata. Creation is a 3-second op; storage is $0 until write. The preview is a full mirror without a copy cost.
• Idempotent CREATE OR REPLACE + DROP IF EXISTS — every re-run is safe. Race conditions between create + push don't corrupt state.
• pull_request:closed trigger — fires on both merge and reject. Reclaims the schema within 30 seconds of PR close.
• Nightly safety-net cron — bounds staleness at 14 days regardless of failure mode. Catches abandoned PRs, GHA outages, fork PRs.
• Cost — $0 storage (zero-copy), ~$0.05 compute per PR review session (XS warehouse, AUTO_SUSPEND=60), $0.01 workflow overhead. Net cost per PR under $0.10. Compared to the "no preview + wait for merge" baseline the analyst-hour savings alone pay for the system in one week.

SQL
Topic — sql
SQL warehouse-preview and clone problems

Practice →

Design
Topic — design
Design problems on ephemeral environments

Practice →

---

Cheat sheet — CI/CD recipes for dbt + Airflow + Spark

• Which gate when. dbt: dbt build --select state:modified+ --defer --state prod-manifest/ --target ci. Airflow: DagBag(dag_folder='airflow/dags', include_examples=False).import_errors == {} first; then operator pytest; then optional dag.test() on LocalExecutor + Postgres service. Spark: local SparkSession.builder.master("local[2]") pytest first, then optional docker-compose bitnami/spark:3.5.1 mini-cluster, then spark-submit --num-executors 0 dry-run against a dev cluster. Preview: CREATE OR REPLACE SCHEMA ANALYTICS.pr_<PR>_<branch> CLONE ANALYTICS.PROD on PR open; DROP SCHEMA IF EXISTS ... CASCADE on PR close; nightly cron drops pr_% older than 14 days.
• Slim CI command template. dbt build --select state:modified+ --defer --state prod-manifest/ --target ci --fail-fast + dbt clone --state prod-manifest/ --resource-type model --resource-type seed --resource-type snapshot. Prod side uploads manifest.json + run_results.json + generated_at.txt to s3://dbt-artifacts/prod/latest/ at the end of every successful run. CI downloads all three; the generated_at.txt freshness guard warns if manifest is >48 h old.
• Path-based short-circuit. Use dorny/paths-filter@v3 at the top of the workflow with dbt: ['dbt/**'], airflow: ['airflow/**'], spark: ['spark/**']. Never run a system's gates on a PR that did not touch that system. The cheapest CI optimisation ever invented.
• DAG parse gate snippet. python -c "from airflow.models.dagbag import DagBag; b=DagBag('airflow/dags', include_examples=False); assert not b.import_errors, b.import_errors" with AIRFLOW_HOME=/tmp/airflow, AIRFLOW__CORE__LOAD_EXAMPLES=False, AIRFLOW__CORE__UNIT_TEST_MODE=True. Runs in 15-30 s; catches every import-time break.
• pytest SparkSession fixture template. @pytest.fixture(scope="session") → SparkSession.builder.master("local[2]").appName("ci").config("spark.sql.shuffle.partitions", "2").config("spark.ui.enabled", "false").config("spark.sql.session.timeZone", "UTC").getOrCreate(). Session-scoped so JVM starts once. Under 200 ms per test after warm.
• Preview schema naming + cleanup. Format pr_<PR>_<slug> where slug = branch|tr '/' '_' | tr -cd '[:alnum:]_' | tr '[:upper:]' '[:lower:]' | cut -c1-40. Create with CREATE OR REPLACE SCHEMA ... CLONE prod; drop with DROP SCHEMA IF EXISTS ... CASCADE. Nightly cron: SELECT SCHEMA_NAME FROM INFORMATION_SCHEMA.SCHEMATA WHERE SCHEMA_NAME LIKE 'PR_%' AND CREATED < DATEADD(day, -14, CURRENT_TIMESTAMP()).
• Deferral vs clone decision matrix. --defer alone: reads unbuilt refs from prod schema; preview schema contains only rebuilt models. Use when the CI job is the only consumer. --defer + dbt clone: preview schema is a full mirror of prod overlaid with rebuilt models. Use when humans (analysts, PMs) query the preview. Cost: identical (both are zero-copy on Snowflake / Databricks UC).
• Cost guardrails. Snowflake: WAREHOUSE_SIZE=XSMALL, AUTO_SUSPEND=60, AUTO_RESUME=TRUE — an XS is $1/hr; 60-s suspend keeps typical CI session under $0.10. Databricks: SQL warehouse auto_stop_mins: 5; interactive cluster autotermination_minutes: 10. BigQuery: on-demand; every query priced independently. Publish per-gate cost budget as a contract (e.g. dbt-slim-ci ≤ $0.30, airflow-parse ≤ $0.02).
• Airflow OCI image versioning contract. Build args → OCI labels: GIT_SHA → org.opencontainers.image.revision, GIT_BRANCH → org.opencontainers.image.ref.name, CI_RUN_ID → ai.pipecode.ci.run.id, AIRFLOW_VERSION → ai.pipecode.airflow.version. Immutable tag sha-<short> for deploys; mutable pr-<PR> / main for humans. docker inspect reveals provenance in one command.
• Spark packaging contract. PySpark: pip wheel → job-<version>-py3-none-any.whl + entry_points = {"console_scripts": [...]}. Scala: sbt assembly → job-<version>-assembly.jar. Config never baked into artifact; selected by env var at submit time. --version flag prints git SHA. --dry-run flag constructs SparkSession + exits 0.
• State-artifact freshness guard. dbt CI downloads generated_at.txt alongside manifest.json; if age_hours > 48 emit a ::warning:: in GitHub Actions. Stale manifest = slim CI over-selects (marks unchanged as modified) or under-selects (misses real changes). Warning surfaces the problem before it silently distorts CI output.
• Full-build fallback triggers. dbt slim CI does not detect all impact classes. Fall back to full build when the PR touches: macros/**, dbt_project.yml, packages.yml, requirements.txt, or profiles.yml. Detect via git diff --name-only origin/main | grep -qE '(macros/|dbt_project\.yml|packages\.yml)'.
• PII-sensitive preview pattern. Don't zero-copy clone PII schemas; instead materialise a 5% SAMPLE with SHA-256 redaction on sensitive columns (SHA2(email, 256) AS email). Storage cost applies (~5% of source); worth it for GDPR / SOC2. Document redaction list in versioned config.
• On-call runbook order. (1) CI red on prod-blocking PR → check gate that failed; (2) dbt slim-CI failed on state comparison → verify s3://dbt-artifacts/prod/latest/manifest.json exists and is <48 h old; (3) Airflow parse failed → open the import_errors traceback; (4) Spark dry-run failed → check dev-EMR cluster health; (5) preview schema missing → check create workflow logs, re-run manually; (6) preview schema orphaned → nightly cron logs at 06:00 UTC.

Frequently asked questions

What is slim CI in dbt?

slim ci dbt is the pattern of running dbt build --select state:modified+ --defer --state prod-manifest/ in CI instead of a full dbt build. The prod manifest.json from the last successful production run is downloaded as an artifact; state:modified+ picks only the models whose SQL, config, or upstream refs have changed since that manifest, plus their descendants; --defer routes any unbuilt parent refs to the prod schema rather than re-materialising them in the CI schema. The result is that a CI run rebuilds 3-15 models on a typical PR instead of 200-1500, cutting the p95 CI time from about 45 minutes to about 2 minutes and the compute cost from $50-$500 per run to about $0.30. Every senior analytics engineer names slim CI in the first minute of any dbt CI/CD interview because it is the load-bearing CI optimisation for dbt projects.

How do I set up state comparison for dbt CI?

State comparison requires an artifact contract between production and CI: the prod dbt job uploads target/manifest.json (plus run_results.json and a generated_at.txt freshness marker) to a known object-store location (typically s3://dbt-artifacts/prod/latest/) at the end of every successful run. The CI workflow downloads those files into a local prod-manifest/ directory before running dbt build; the --state prod-manifest/ flag tells dbt where to look. Under the hood, dbt fingerprints every model (SHA-256 of SQL text + config + upstream refs + sources + contract), compares the CI manifest against the prod manifest, and marks any model with a differing fingerprint as state:modified. Ship the freshness guard alongside — warn if the prod manifest is more than 48 hours old, because stale prod state distorts the diff (over-selects or under-selects). Bootstrap by running one manual full build against the target-prod schema and uploading its manifest; subsequent runs incrementally refresh the artifact.

Do I need a full Spark cluster in CI?

No, and running one is the single most common CI-cost mistake in Spark shops. The CI matrix for Spark is three gates in ascending cost: (a) local SparkSession.builder.master("local[2]") pytest fixture, which runs in the GHA runner with zero cluster cost and covers pure transformations, UDFs, and schema assertions — this catches roughly 80% of Spark bugs; (b) docker-compose bitnami/spark:3.5.1 mini-cluster for jobs that exercise real shuffle, broadcast joins, or partitioning behaviour — adds ~2 minutes to CI when it runs but stays gated on spark/jobs/** path filter; (c) a spark-submit --num-executors 0 --dry-run submit against a small dev EMR cluster to validate config, JAR resolution, and cluster reachability without ever provisioning executors. The real prod cluster is never touched by CI. If you find your CI job submitting to the prod Spark cluster, you have a cost bug — refactor to the three-gate pattern.

How do I test Airflow DAGs in CI?

Three layers of gates in ascending cost. First, the DAG parse gate: instantiate DagBag(dag_folder='airflow/dags', include_examples=False) in CI and assert dagbag.import_errors == {}. Runs in 15-30 seconds and catches every import-time break — the single highest-impact class of Airflow bug because import failures pause the whole DagBag, not just the broken DAG. Second, pytest against custom operators and hooks with mocked backends: instantiate the operator, provide a mock context, assert on return values and mock call args. Coverage bar: 95% line coverage on everything in airflow/plugins/. Third, an optional integration test using dag.test(execution_date=...) (Airflow 2.5+) with a LocalExecutor and a Postgres service — runs the entire DAG synchronously without a scheduler. Reserve the integration layer for high-value DAGs (ones touching shared warehouse tables, dynamic task mapping, etc.) so the CI matrix stays under 6 minutes. Wrap all three in a labelled OCI image built with apache/airflow:2.10.x as the deployable artifact.

How do preview environments work with Snowflake?

Every open pull request maps to a throwaway schema named pr_<PR>_<branch> in Snowflake, created on PR open via zero-copy clone and dropped on PR close. The create workflow runs CREATE OR REPLACE SCHEMA ANALYTICS.pr_4712_add_orders_v2 CLONE ANALYTICS.PROD — Snowflake implements clone as a metadata operation, so a 10 TB source schema clones in about 3 seconds and costs $0 in storage until a write happens. The slim-CI dbt build runs against the preview schema; dbt clone fills in any unmodified tables so the preview is a full mirror. A comment is posted to the PR with a Snowsight deep-link that pre-selects the schema, so reviewers open the preview in one click. The drop workflow triggers on pull_request type closed (both merge and reject) and runs DROP SCHEMA IF EXISTS pr_<PR>_<branch> CASCADE. A nightly cron at 06:00 UTC drops any pr_% schema older than 14 days as a safety net for abandoned PRs or GHA outages. Compute uses an XS warehouse (WAREHOUSE_SIZE=XSMALL, AUTO_SUSPEND=60), keeping typical per-PR compute cost under $0.10.

What is dbt deferral vs dbt clone?

dbt deferral (--defer --state prod-manifest/) is a read-side optimisation: when a CI-built model runs {{ ref('unmodified_upstream') }}, dbt sees the ref was not rebuilt in this run, consults the state manifest, and rewrites the ref to point at the prod schema instead of the CI schema. No table is materialised for the deferred parent in the preview schema; the read is redirected. dbt clone (dbt clone --state prod-manifest/ --resource-type model --resource-type seed --resource-type snapshot) is a write-side operation: for every unmodified model in the state manifest, dbt executes CREATE OR REPLACE <preview>.<model> CLONE prod.<model>, producing a zero-copy metadata pointer in the preview schema. Use --defer alone when the CI job is the only consumer (fast, no schema footprint). Use --defer + dbt clone when humans (analysts, PMs) query the preview and expect all the joined tables to be there. Cost is identical (both are zero-copy on Snowflake and Databricks UC); the difference is whether the preview schema is a full mirror or just a delta.

Practice on PipeCode

• Drill the ETL practice library → for the pipeline-CI, slim-build, and state-comparison problems senior interviewers love.
• Rehearse on the SQL practice library → for warehouse-side clone semantics, incremental modeling, and preview-schema query patterns.
• Sharpen the systems axis with the design practice library → for CI/CD architecture, ephemeral environment design, and deployable-artifact provenance scenarios.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the four-axis CI decision matrix against real graded inputs.

Lock in pipeline-CI muscle memory

Docs explain slim CI. PipeCode drills explain the decision — when `state:modified+` is enough versus when to fall back to a full build, when the DAG parse gate saves the scheduler versus when a `dag.test()` integration is worth the six minutes, when a docker-compose Spark cluster is worth the setup cost versus when `local[2]` pytest is the honest answer, when a zero-copy preview schema replaces "wait for merge" as the analyst review UX. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the CI/CD trade-offs senior analytics engineers and data platform engineers actually face.

Practice ETL problems →
Practice design problems →