Module 0 — Environment & Foundations

n8n Data Platform Engineer — Skills Track

Local-first setup connecting a self-hosted automation tool to a cloud data warehouse, an ELT transformation layer, and an orchestration layer — the four systems this track builds on.

---

Architecture

flowchart LR
    subgraph Local["🖥️ Local VM — Ubuntu"]
        N8N["🔗 n8n<br/>workflow automation<br/>:5678"]
        DBT["🏗️ dbt-core<br/>transformation layer"]
        DAG["🎛️ Dagster<br/>orchestration<br/>:3000"]
    end

    subgraph Cloud["☁️ Google Cloud"]
        BQ["🗄️ BigQuery<br/>n8n-self-practice<br/>sandbox tier"]
    end

    N8N -.future integration.-> BQ
    DBT -->|oauth auth| BQ
    DAG -->|will orchestrate| DBT

    style N8N fill:#EA4B71,stroke:#333,color:#fff
    style DBT fill:#FF694B,stroke:#333,color:#fff
    style DAG fill:#6E43E8,stroke:#333,color:#fff
    style BQ fill:#4285F4,stroke:#333,color:#fff
    style Local fill:#F8F9FA,stroke:#333,stroke-width:2px
    style Cloud fill:#E8F0FE,stroke:#1a73e8,stroke-width:2px

Loading

---

What's Working

Component	Status	Verified By
🔗 n8n	✅ Running	Local Docker, localhost:5678 reachable
☁️ BigQuery auth	✅ Connected	gcloud auth application-default login
🏗️ dbt → BigQuery	✅ Connected	dbt debug → All checks passed
🎛️ Dagster	✅ Running	localhost:3000 webserver live

---

Setup Flow

sequenceDiagram
    participant Dev as 👩‍💻 Developer
    participant GCP as ☁️ Google Cloud
    participant dbt as 🏗️ dbt-core
    participant Dag as 🎛️ Dagster

    Dev->>GCP: gcloud init (select project)
    Dev->>GCP: gcloud auth application-default login
    Dev->>dbt: dbt init my_project (oauth method)
    dbt->>GCP: dbt debug (verify connection)
    GCP-->>dbt: ✅ All checks passed
    Dev->>Dag: dagster project scaffold
    Dev->>Dag: dagster dev

Loading

1. gcloud init — authenticate CLI, select GCP project (n8n-self-practice)
2. gcloud auth application-default login — application-level credentials for dbt
3. Python venv created outside the dbt project folder (~/n8n-practice/venv)
4. dbt init my_project — oauth authentication method, no service account key needed (sandbox tier constraint)
5. dbt debug — confirms warehouse connection
6. dagster project scaffold + dagster dev — orchestration layer running, not yet wired to any real asset

---

🔍 Self-Check: Tested vs Assumed

✅ Tested:

• dbt connects to BigQuery and passes all debug checks
• n8n workflow persists after container restart (Docker volume mount confirmed)
• Dagster webserver serves UI on localhost:3000
• Service account/OAuth scope limited to sandbox project — no billing account attached

⚠️ Assumed (not yet tested):

• Behavior once real data volume exceeds sandbox free-tier limits
• Dagster-to-dbt integration (dagster-dbt) — install confirmed, integration not yet wired
• n8n-to-BigQuery direct integration — not yet attempted

---

➡️ Next Module

Module 1 — Advanced SQL for Analytics Engineering — window functions, CTEs, and incremental-query patterns against a public BigQuery dataset.

---

Module 1 — Advanced SQL for Analytics Engineering

Practiced against bigquery-public-data.thelook_ecommerce.orders — a public e-commerce dataset. Full queries in sql/module1_advanced_sql.sql.

Concepts Covered

mindmap
  root((Advanced SQL))
    Window Functions
      COUNT OVER PARTITION
      Per-user sequencing
    Running Totals
      SUM OVER
      ROWS BETWEEN
    Deduplication
      ROW_NUMBER
      QUALIFY filter
    Aggregation
      Monthly cohorts
      GROUP BY and EXTRACT

Loading

Query Summary

#	Concept	What It Solves
1	Basic exploration	Understand raw table shape and columns
2	Window function COUNT OVER	Number each user's orders in sequence
3	Running total SUM OVER	Cumulative items ordered per user, over time
4	Dedup ROW_NUMBER plus QUALIFY	Isolate latest order per user, reused in dbt staging models
5	Monthly aggregation	Cohort-style breakdown by year, month, status

Self-Check: Tested vs Assumed

Tested:

• All 5 queries return correct, verified results against the public dataset
• QUALIFY correctly filters window function output, not possible with plain WHERE
• Running total values manually spot-checked against raw num_of_item values

Assumed, not yet tested:

• Query performance and cost at production-scale row counts, millions plus rows
• Behavior with NULL values in partition or order columns, not encountered in this sample

Next Module

Module 2 — dbt Fundamentals on BigQuery — staging and marts models, first working ELT pipeline.

---

Module 2 — dbt Fundamentals on BigQuery

First working ELT pipeline: raw BigQuery public data transformed through staging into a business-ready mart table.

Pipeline Lineage

Models Built

Layer	Model	Purpose
Staging	stg_orders	Cleaned order records from raw source
Staging	stg_users	Cleaned user records from raw source
Staging	stg_order_items	Cleaned order line items from raw source
Mart	fct_orders	Joined, business-ready order fact table with total value

Self-Check: Tested vs Assumed

Tested:

• All 3 staging models build successfully as views
• fct_orders correctly joins staging models via ref, dbt-managed dependency graph
• 13 of 13 data tests pass, unique and not_null on key columns
• Lineage graph confirms correct dependency order

Assumed, not yet tested:

• Incremental materialization, all models are full-refresh views so far
• Behavior on schema drift if source table columns change

Next Module

Module 3 — Data Modeling: Dimensional Design — star schema and SCD Type 2 snapshots.

---

Module 3 — Data Modeling: Dimensional Design

Star schema built on top of Module 2's staging layer, with SCD Type 2 change tracking via dbt snapshots.

Star Schema

erDiagram
    fct_orders }o--|| dim_users : user_key
    fct_orders }o--|| dim_date : order_date_key
    fct_order_items }o--|| dim_users : user_key
    fct_order_items }o--|| dim_products : product_key
    fct_order_items }o--|| dim_date : order_date_key
    fct_order_items }o--|| fct_orders : order_id

    dim_users {
        string user_key PK
        string user_id
        string city
        boolean is_current
    }
    dim_products {
        string product_key PK
        string product_id
        string category
    }
    dim_date {
        date date_key PK
    }
    fct_orders {
        string order_id PK
        string user_key FK
        date order_date_key FK
    }
    fct_order_items {
        string order_item_id PK
        string user_key FK
        string product_key FK
        date order_date_key FK
    }

Loading

Models Built

Layer	Model	Purpose
Snapshot	users_snapshot	SCD Type 2 change tracking on users (check strategy)
Dimension	dim_users	Surrogate key, current + historical user attributes
Dimension	dim_products	Surrogate key, static product attributes
Dimension	dim_date	Calendar spine, 2019–2026
Fact	fct_orders	Order grain — 1 row per order
Fact	fct_order_items	Line-item grain — 1 row per order item, carries product_key

Design Decisions

• fct_orders vs fct_order_items split: an order can contain multiple products, so product_key cannot live on the order-grain fact without breaking grain. Product-level analysis goes through fct_order_items instead.
• SCD2 fan-out guard: joins from fact tables to dim_users filter on is_current = true to prevent duplicate rows if a user's tracked attributes ever change.
• check strategy over timestamp: source table has no updated_at column, so snapshot uses check_cols on the fields most likely to change (address, traffic_source) rather than a blanket column check.

Self-Check: Tested vs Assumed

Tested:

• Snapshot correctly captures dbt_valid_from/dbt_valid_to on first run
• Found and fixed a source data-quality issue — city column held the literal string "null" for ~5% of Brazil rows, silently bypassing IS NULL checks
• All fact-to-dimension joins verified for null foreign keys (0 nulls on user_key, product_key, date_key)
• 19 of 19 data tests pass across staging, dimensions, and facts

Assumed, not yet tested:

• SCD2 behavior on an actual attribute change — source data is static, so a second snapshot run producing a new row version hasn't been observed yet
• dim_date coverage beyond 2026 if the source data range extends further

Next Module

Module 4 — Orchestration with Dagster

The Module 2/3 dbt project wrapped in a Dagster orchestration layer: scheduled runs, freshness observability, and real Slack alerting on failure.

Architecture

flowchart LR
    subgraph dbt["🏗️ dbt Project"]
        M["9 models<br/>28 tests"]
    end

    subgraph Dagster["🎛️ Dagster Orchestration"]
        A["📦 dbt Assets<br/>dagster-dbt wrap"]
        S["⏰ Schedule<br/>daily 06:00 UTC"]
        F["📈 FreshnessPolicy<br/>26h tolerance"]
        RS["🚨 run_failure_sensor"]
    end

    subgraph Alert["🔔 Alerting"]
        SL["💬 Slack #new-channel"]
    end

    M -->|wrapped as| A
    S -->|triggers| A
    A -->|tracked by| F
    A -->|on FAILURE| RS
    RS -->|POST webhook| SL

    style dbt fill:#FF694B,stroke:#333,color:#fff
    style Dagster fill:#F3F0FF,stroke:#6E43E8,stroke-width:2px
    style Alert fill:#ECE4F7,stroke:#4A154B,stroke-width:2px
    style A fill:#6E43E8,stroke:#333,color:#fff
    style S fill:#8B6EF0,stroke:#333,color:#fff
    style F fill:#8B6EF0,stroke:#333,color:#fff
    style RS fill:#E8433E,stroke:#333,color:#fff
    style SL fill:#4A154B,stroke:#333,color:#fff

Loading

Induced-Failure Self-Check

sequenceDiagram
    participant Dev as 👩‍💻 Developer
    participant Model as 🏗️ stg_orders.sql
    participant Dag as 🎛️ Dagster
    participant BQ as ☁️ BigQuery
    participant Slack as 💬 Slack

    Dev->>Model: typo order_id → order_idd
    Dev->>Dag: Materialize all 9 assets
    Dag->>BQ: run stg_orders
    BQ-->>Dag: ❌ column not found
    Dag-->>Dag: fct_orders, fct_order_items fail (propagation)
    Dag->>Slack: 🚨 run_failure_sensor fires
    Slack-->>Dev: "dbt run failed: run_id=8aab9e24"
    Dev->>Model: revert typo
    Dev->>Dag: Materialize again
    Dag->>BQ: run all 9 models
    BQ-->>Dag: ✅ all pass

Loading

What's Working

Component	Status	Verified By
🎛️ dagster-dbt integration	✅ Working	All 9 assets materialize via Dagster, not raw dbt
⏰ Daily schedule	✅ Enabled	Toggled ON in UI, materialize_dbt_models_schedule
📈 Freshness observability	✅ Working	Asset detail page shows "Passing" state
🚨 Slack alert on failure	✅ Working	Real message received: run_id 8aab9e24
🔁 Failure → fix → pass cycle	✅ Verified	Run 8aab9e24 (fail) → Run 94bd25d5 (pass)

🔍 Self-Check: Tested vs Assumed

✅ Tested:

• dbt project wrapped as native Dagster assets via dagster-dbt, all 9 models visible and materializable in UI
• Daily schedule defined and enabled (cron 0 6 * * *)
• Freshness tracked via FreshnessPolicy (new API) — confirmed "Passing" state on a live asset after materialization
• Induced failure (typo'd column) correctly failed stg_orders and cascaded to fct_orders/fct_order_items, while unrelated assets stayed green
• run_failure_sensor fired a real Slack webhook message on failure, confirmed in-channel
• Revert-and-repass confirmed with a second clean run (paired run IDs as evidence)
• Concurrent-run behavior tested (not assumed) — two back-to-back materialize triggers exposed a real race condition: DbtProject.prepare_if_dev() running twice simultaneously briefly left dbt_packages/ in an inconsistent state, failing one run with "0 package(s) installed." This happened despite dagster_dbt's own run_with_concurrent_update_guard — a genuine limitation, not a VM issue

⚠️ Assumed (not yet tested):

• Backfill behavior across multiple missed schedule windows — not triggered in this project
• Fix for the concurrent-run race condition above — documented as a known limitation, not patched (out of scope for this module)

Notes on the Build

• dagster-dbt's factory functions (build_last_update_freshness_checks, build_sensor_for_freshness_checks) turned out to be superseded by a newer FreshnessPolicy/apply_freshness_policy API in this Dagster version (1.13.15) — migrated to the new pattern after hitting a SupersessionWarning.
• Both the dbt CLI resource and the project's dbt-project object needed an explicit profiles directory configured — the dbt profile lives outside the project folder in this setup.
• Two local dagster dev crashes during testing were traced to local machine memory pressure (swap near capacity, limited available RAM) rather than an application defect — resolved by restarting; a permanent fix (increasing available memory) is still pending.

Next Module

Module 5 — Gemini-Enrichment Ingestion Pipeline

A custom ingestion pipeline that pulls unprocessed records from BigQuery, enriches them via the Gemini API (structured JSON output), validates the response, dedups against prior runs, and lands results in BigQuery — orchestrated as a Dagster asset upstream of the Module 2/3 dbt project.

Architecture

flowchart LR
    A[raw.pending_enrichment] -->|fetch rows| B[gemini_enriched_records<br/>Dagster asset - main venv]
    B -->|JSON via stdin| C[gemini_batch_runner.py<br/>subprocess - venv-gemini]
    C -->|call_gemini| D[Gemini API<br/>gemini-3.5-flash]
    D -->|structured JSON| C
    C -->|validate + dedup| E{Valid and new?}
    E -->|yes| F[raw.gemini_enriched]
    E -->|no| G[raw.failed_llm_calls]
    C -->|JSON via stdout| B
    B -->|metadata + failure-rate check| H[Dagster UI]

Loading

Why a subprocess bridge, not a direct import: the Gemini SDK (google-genai) requires protobuf>=3.20,<6, while dbt-core in this project requires protobuf>=6. Both packages can't coexist in one virtual environment - verified directly (not assumed) by hitting a real pip dependency-resolution conflict. Fix: two separate venvs (venv for dbt/Dagster, venv-gemini for the Gemini SDK), bridged via subprocess.run() passing records as JSON over stdin/stdout.

Sequence - one record through the pipeline

sequenceDiagram
    participant D as Dagster asset (main venv)
    participant R as gemini_batch_runner.py (venv-gemini)
    participant G as Gemini API
    participant BQ as BigQuery

    D->>BQ: SELECT id, text FROM raw.pending_enrichment
    D->>R: subprocess.run(records as JSON via stdin)
    R->>G: generate_content(response_schema=EnrichmentResult)
    G-->>R: structured JSON
    R->>R: Pydantic validate
    alt valid and new
        R->>BQ: INSERT INTO raw.gemini_enriched
    else invalid or duplicate
        R->>BQ: INSERT INTO raw.failed_llm_calls (or skip)
    end
    R-->>D: summary JSON via stdout
    D->>D: add_output_metadata + failure-rate check

Loading

What's Working

Component	Status	Verified By
Gemini API auth	Working	Live call, key confirmed via client.models.generate_content
Structured output (response_schema)	Working	Real Gemini response parsed into EnrichmentResult without manual JSON parsing
Idempotent dedup	Working	Same input_id/text re-run returned skipped_duplicate, no duplicate row
Retry/backoff (transient errors)	Working	tenacity exponential backoff wired for 429/timeout/503/UNAVAILABLE
Fatal-error handling (batch-wide)	Working	Invalid API key correctly raised GeminiFatalError, halted process_batch cleanly
Schema validation	Working	Two malformed-response cases both routed to raw.failed_llm_calls
Cross-venv subprocess bridge	Working	Standalone test and full Dagster UI materialize both succeeded
Dagster asset (gemini_enriched_records)	Working	Materialized via UI, metadata shows inserted=3, skipped_duplicate=0, failed_validation=0, failed_api=0

Self-Check: Tested vs Assumed

Tested:

• Deprecated SDK caught live, not assumed - google.generativeai raised a FutureWarning and gemini-1.5-flash 404'd; migrated to google-genai SDK with gemini-3.5-flash, re-verified with a live auth call
• Dedup key is sha256(input_id + text), NOT the Gemini output - deliberately, since Gemini's output is non-deterministic across calls
• Auth/permission/quota errors are treated as fatal (halt the whole batch), while per-record content issues are logged individually - confirmed both paths behave differently via separate induced tests
• Real end-to-end run through the actual Dagster UI processed 3 real records with zero errors

Assumed (not yet tested):

• Behavior at production call volume / real Gemini rate-limit thresholds - only tested with a 3-record batch
• Concurrent Dagster runs of this asset specifically (Module 4 found a real race condition in the dbt asset under concurrent materialization)
• Cost per run at scale - not yet calculated or budgeted

Notes on the Build

• dbt-core requires protobuf>=6, the Gemini SDK requires protobuf<6 - a hard, unresolvable conflict in one venv, not a version-pinning issue. Two venvs + subprocess bridge was the correct fix.
• The subprocess contract keeps stdout reserved for a single JSON object (the summary); all logging is redirected to stderr in gemini_batch_runner.py.
• gemini_ingestion.py calls load_dotenv() explicitly rather than relying on shell environment, since it now runs both as a directly-imported module and as a subprocess invoked by Dagster with its own working directory.

Module 6

Module 6 — CI/CD & Pipeline Security Hygiene

Overview

This module adds automated pull-request testing (slim CI) and hardens the pipeline's cloud access model — moving from a single broad-access dev identity to a least-privilege, keyless, per-PR-isolated CI setup.

Architecture

flowchart TD
    A[Push to main] --> B[store-manifest.yml]
    B --> C[Authenticate via Workload Identity Federation]
    C --> D["dbt build --target ci (full build)"]
    D --> E[Upload manifest.json as artifact]

    F[Open Pull Request] --> G[dbt-ci.yml]
    G --> C
    G --> H[Download main's manifest artifact]
    H --> I["dbt build --select state:modified+ --defer, schema=dbt_ci_pr<N>"]
    I --> J{Pass?}
    J -->|Yes| K[Merge allowed]
    J -->|No| L[Merge blocked]
    I --> M[Cleanup: delete temp per-PR dataset]

Loading

Security Model

• No long-lived keys: GitHub Actions authenticates to GCP via Workload Identity Federation (OIDC). No key file exists anywhere in this repo or on disk.

• Repo-scoped trust: the WIF provider's attribute condition restricts token exchange to this exact repo — no other GitHub repo can impersonate the CI identity.

• Least-privilege CI service account (dbt-ci-runner):

  Resource	Access
  dbt_ci dataset	WRITER
  raw dataset	READER
  snapshots dataset	WRITER (corrected after real failure — see Notes)
  dbt_dev dataset	No access
  Custom role dbtCiDatasetCreator	bigquery.datasets.create only, for per-PR schemas

• Branch protection on main: requires a passing dbt Slim CI status check, enforced for repo admins too (Do not allow bypassing). Approval requirement was disabled — GitHub blocks self-approval, and this is a solo-developer repo, so the status check is the enforced gate, not review.

Slim CI + Per-PR Isolation

PRs rebuild only what changed (state:modified+), in an isolated temporary schema (dbt_ci_pr<N>) via a custom generate_schema_name macro, deferring unmodified models to main's full baseline (--defer). The temporary schema is deleted after each run.

Self-Check: Tested vs. Assumed

Behavior	Status
WIF auth works end-to-end from GitHub Actions to GCP	Tested
Slim CI runs and passes on a real PR	Tested
CI correctly fails and blocks merge on broken SQL	Tested — intentional syntax error caused a failing check and blocked merge
No secret ever committed to git history	Tested — full history grep for key/secret patterns, empty
Branch protection blocks direct push, including for the repo owner	Tested — initially found admins were exempted by default; fixed and reconfirmed with a real rejected push
Two PRs running CI at the same time don't corrupt each other's data	Tested — first attempt genuinely failed (shared dbt_ci dataset race, same class of bug as Module 4's dbt_packages race); fixed with per-PR schema isolation + --defer, then re-tested and confirmed passing
CI service account has correct permissions for every dbt operation it performs	Tested — first attempt under-scoped snapshots to read-only, which broke dbt build's snapshot step; corrected to WRITER after real failure, not assumption
Behavior under 3+ simultaneous PRs, or PRs touching the same model	Not tested — only a 2-PR, non-overlapping-model scenario was verified

Notes on the Build

• GCP IAM policy binding at the dataset level required the classic bq ACL (JSON) method instead of bq add-iam-policy-binding, since this sandbox project returned a "requires allowlisting" error for the newer API.
• Workload Identity Federation was chosen over a JSON key despite being more complex to set up, to avoid any long-lived credential existing at all.
• The first version of store-manifest.yml used dbt compile (SQL generation only, no tables created), which silently meant the baseline dataset never had a full build — this only surfaced under the concurrency test, not from local testing.
• The CI service account was initially scoped read-only on snapshots, which broke dbt's snapshot write step — corrected after a real failure.

Module 7 — LLM Pipeline Hardening

This module hardens the Gemini enrichment pipeline built in Module 5 -- adding cost/latency controls, a human-review gate for low-confidence output, adversarial input test coverage, and a formal threat model. Not a rebuild of Module 5; a security/reliability layer on top of it.

Architecture

Flow inside gemini_ingestion.py (runs in venv-gemini):

process_record(record, tracker)
  -> tracker.check_before_call()   # blocks BEFORE any API spend if budget exhausted
  -> call_gemini(record)           # returns (raw, input_tokens, output_tokens)
  -> tracker.record_call(...)      # logs cost/latency to raw.llm_cost_log
  -> if confidence < 0.7:
       write_to_review_queue()     # raw.hitl_review_queue
     else:
       write_success()             # raw.gemini_enriched (existing Module 5 path)

cost_tracker.py is duplicated in both my_project/ (imported by gemini_ingestion.py, venv-gemini) and orchestration/orchestration/ (available to the Dagster asset's venv) -- kept manually in sync, not a package, since the two venvs can't share an editable install without reintroducing the protobuf conflict Module 5 already worked around.

Self-Check: Tested vs. Assumed

Behavior	Status
Cost/latency logged per Gemini call to raw.llm_cost_log	Tested -- real row confirmed via bq query, correct token counts and cost
Budget ceiling blocks the NEXT API call before it happens, once exhausted	Tested -- forced via manual tracker.total_cost_usd override, confirmed BudgetExceededError raised, no API call made
Low-confidence output routed to raw.hitl_review_queue instead of gemini_enriched	Tested -- forced via monkeypatched call_gemini returning confidence 0.3, confirmed queued_for_review status and real row insert
Gemini's self-reported confidence score is a reliable signal	Not assumed reliable -- observed high confidence on deliberately garbage input during testing; documented as a known limitation, not fixed
Adversarial inputs (prompt injection, oversized text, SQL-looking strings) handled safely	Tested -- 7 real tests against live process_record, all passing. One run hit a free-tier rate limit (429, 5 req/min) which raised GeminiFatalError as designed -- confirmed as a rate-limit, not a logic bug, by retrying after cooldown
raw.failed_llm_calls.raw_response redacted before storage	Tested -- regex-based redact_pii() for emails/phones, wired into write_failure(), confirmed via real schema-validation-failure path. Partial mitigation only -- does not catch names, addresses, or other free-text PII
GEMINI_API_KEY exposure re-verified for this module's new code paths	Not re-verified -- relies on Module 5's original .env/gitignore check

Notes on the Build

• The first patch script (apply_module7_patch.py) placed its execution loop before all edits were appended to the list -- the process_record/process_batch edits added afterward never actually ran, despite the script printing "Applied 3/3 edits". Caught by testing the new tracker parameter directly (TypeError: unexpected keyword argument), not assumed working from a clean exit code. Fixed with a second patch script.
• cost_tracker.py's summary() originally rounded to 4 decimals, which displayed real sub-cent costs (e.g. $0.0000114) as 0.0. Data in BigQuery was correct the whole time -- this was a display bug, caught by cross-checking the printed summary against the actual bq query row.
• Token count field names (prompt_token_count, candidates_token_count) were verified against a live response.usage_metadata object before trusting them, rather than assumed from general google-genai familiarity.

Next Module

Module 8 — TBD