Lineage, Contracts, and Observability

The finale: three views of one lineage graph, dbt contracts that let you change a mart on a Friday, and observability that pages for a broken warehouse but not a warned test.

Everything in this series has been about assembling the stack. This last post is about trusting it — which is a different problem. A pipeline you can’t see into is a pipeline you’re afraid to change, and a stack you’re afraid to change slowly ossifies until the only safe move is no move. The three things that break that fear are the three subjects here: lineage so you know what depends on what, contracts so you can change one link without guessing about the rest, and observability so you find out a run went wrong before a stakeholder does.

Each of the three tools you’ve bolted together holds a fragment of the truth. dbt knows how models and columns depend on each other. Airflow knows how tasks depend on each other. Snowflake knows who actually read which table. None of them, alone, can answer the question you’ll be asked in an incident — if I change raw.orders, what breaks all the way to the BI dashboard? — because the answer runs through all three. So we start by making the three fragments into one picture.

Three views of one graph

dbt’s view is the transformation graph, down to the column. Because every model refers to its parents through ref() and source(), dbt already knows the edges — that’s how Cosmos turned your project into an Airflow DAG in the first place. What’s easy to miss is that dbt also parses the compiled SQL of each model, so it knows which columns flow into which. Run:

dbt docs generate

and dbt writes a manifest.json and catalog.json that together describe the model graph and, where it can resolve them, column-level edges: revenue_by_region.revenue is derived from stg_lineitem.extended_price and stg_lineitem.discount. dbt docs serve renders that as the familiar lineage graph; dbt Explorer (the hosted version) draws the same graph with column-level lineage you can click through — pick stg_orders.customer_id and it highlights every downstream column that consumed it. This view is authoritative for everything inside the warehouse transformation, and blind to everything outside it. It does not know a schedule fired the run, and it does not know a Tableau extract read the mart afterward.

Airflow’s view is the run graph, and OpenLineage is how it leaves the box. Airflow knows the task DAG, but a task DAG is per-pipeline — it can’t tell you that this transform DAG consumes a dataset that ingest DAG produced. OpenLineage is the open standard that fixes this: it models runs, jobs, and the datasets between them, and every tool that speaks it emits events into a shared graph. Airflow ships a first-party provider for it:

pip install apache-airflow-providers-openlineage
# airflow.cfg  (or AIRFLOW__OPENLINEAGE__* env vars)
[openlineage]
transport = {"type": "http", "url": "http://marquez:5000", "endpoint": "api/v1/lineage"}
namespace = tpch_prod

With that in place, every task run emits a start and complete event describing what it read and wrote. Marquez is the reference OpenLineage store and UI — point the transport at it and you get a lineage browser fed by the runs themselves, not by a static parse.

The important part is what Cosmos adds. Because Cosmos renders each dbt model as its own task and runs it through dbt, it can emit per-model OpenLineage events — dbt already knows the input and output relations of every model, so Cosmos attaches that dataset lineage to each task’s event. The result is that the Airflow graph and the dbt graph land in the same Marquez namespace: the ingest DAG’s write to raw.orders, the transform DAG’s models reading it, and the marts they produce all become one connected dataset graph, stitched by the physical Snowflake table names both sides reference. This is the seam that dbt-alone and Airflow-alone can’t cross — dbt doesn’t know about ingest, Airflow doesn’t know about columns, and OpenLineage is where their two graphs meet.

Snowflake’s view is who actually touched the data. dbt lineage is what should read a table; ACCESS_HISTORY is what did. It’s an account-usage view that records, per query, every object read and written — including reads dbt never orchestrated, like an analyst’s ad-hoc query or a BI tool’s nightly extract:

select
    query_id,
    user_name,
    directsource.value:objectName::string as object_read
from snowflake.account_usage.access_history,
     lateral flatten(input => base_objects_accessed) as directsource
where directsource.value:objectName::string
      = 'ANALYTICS.MARTS.REVENUE_BY_REGION'
  and query_start_time > dateadd('day', -7, current_timestamp())
order by query_start_time desc;

base_objects_accessed resolves through views to the underlying tables (as opposed to direct_objects_accessed, which stops at the object named in the query), and each entry carries the columns touched — so you can ask not just who read revenue_by_region but which columns of it anyone read all week. A column no downstream consumer has selected in ninety days is a column you can probably drop. This is the one view of the three that sees past the mart into the humans and tools consuming it.

Combining them for impact analysis. Now the incident question has an answer with a shape. If I change raw.orders, what breaks all the way to the dashboard?

  • dbt lineage walks the inside: source('tpch','orders') → stg_orders → orders_enriched and revenue_by_region. Column-level lineage narrows it — if you’re only dropping o_comment, nothing downstream selected it, and the blast radius is zero.
  • OpenLineage / Marquez walks the run graph: which DAGs produce and consume those datasets, so you know the ingest job upstream and every transform run that would need re-running.
  • ACCESS_HISTORY walks the outside: who queried the affected marts and which columns, so you know which dashboards and which people to warn before, not after.

Concretely, once you’ve decided the change touches revenue_by_region, the “who do I warn” half is one query against the outside view:

select distinct user_name, count(*) as reads_last_30d
from snowflake.account_usage.access_history,
     lateral flatten(input => base_objects_accessed) as o
where o.value:objectName::string = 'ANALYTICS.MARTS.REVENUE_BY_REGION'
  and query_start_time > dateadd('day', -30, current_timestamp())
group by user_name
order by reads_last_30d desc;

The BI_SERVICE account at the top of that list is the dashboard; the human names below it are the analysts who built reports on the mart. That’s your notification list, generated from what actually happened rather than a wiki page someone forgot to update. No single tool spans raw arrival to BI consumer. The three together do, and once they’re all emitting into places you can query — manifest.json, Marquez, ACCOUNT_USAGE — impact analysis stops being a memory exercise and becomes a join.

Contracts: change one part on a Friday

Lineage tells you what you’re about to break. A contract is what lets you not break it. The scenario contracts are built for is the one every data team fears: a mart that BI reads directly, a change that has to ship, and no way to know if the change will silently reshape the table the dashboard is bound to. Without a contract, you find out when the dashboard errors. With one, you find out at build time, in the PR.

dbt model contracts enforce the shape of a model — its columns, their types, and their constraints — as a condition of building it. Declare one on the mart BI depends on:

# models/marts/_marts__models.yml
models:
  - name: revenue_by_region
    access: public
    config:
      contract:
        enforced: true
    columns:
      - name: region
        data_type: varchar
        constraints:
          - type: not_null
      - name: revenue
        data_type: number(38,2)
        constraints:
          - type: not_null

With contract: {enforced: true}, dbt refuses to build the model unless the SELECT produces exactly these columns with these types. Rename revenue to total_revenue, change it from a number(38,2) to a float, or drop the region column, and the build fails with an explicit diff — before the model reaches the warehouse the dashboard reads. The error is specific: dbt names the mismatched column and prints the expected type against the one your SQL produced, so a contract break is a two-line fix in review, not a forensic exercise after a dashboard errors. On Snowflake, dbt enforces this by building against the declared column definitions, so the physical table is guaranteed to match the contract or the run stops. The contract turns “did I just change the interface” from a question you answer by watching the dashboard into one the build answers for you.

There’s a discipline cost worth naming: an enforced contract means every column and type is now declared in YAML, so a genuine, intended change is two edits — the SQL and the contract — instead of one. That friction is the feature. It’s exactly the pause you want before the shape of a public table moves, and it’s cheap precisely on the models where it matters, because a mart BI reads has a handful of columns, not forty. Enforce contracts on the public marts; leave the private staging churn unencumbered.

access marks which models are interfaces at all. A contract protects a table’s shape; access protects which tables outsiders are even allowed to depend on. Three levels:

  • private — only models in the same group may ref() it. Your staging models should be private; nothing outside the transform project has any business joining to stg_orders.
  • protected — any model in the same dbt project may reference it (the default).
  • public — anyone, including other projects and BI tools, may depend on it. This is the promise: this is a stable interface, we won’t reshape it under you.

revenue_by_region is public because BI reads it; the staging models that feed it are private. Marking that distinction is what lets you refactor forty staging models freely while treating the two public marts as contracts you honor.

groups give the private/public wall an owner. A group is a named bundle of models with an owner, and access: private is scoped to a group. Declare one:

# models/_groups.yml
groups:
  - name: finance
    owner:
      name: Analytics Engineering
      email: [email protected]
# then, on the models
models:
  - name: stg_orders
    group: finance
    access: private     # only other `finance` models may ref it
  - name: revenue_by_region
    group: finance
    access: public

Now stg_orders is private to the finance group — a model in some other group that tries to ref('stg_orders') fails to compile, with dbt naming the group boundary it crossed. Groups turn “please don’t depend on our internal staging tables” from a code-review plea into a rule the parser enforces.

versions let a breaking change ship without a flag day. Sometimes the interface genuinely has to change — a new grain, a dropped column a consumer still uses. Contracts would (correctly) block it, and access: public means you can’t just break consumers. Versioning is the escape hatch: publish the new shape as v2 while v1 keeps serving until everyone migrates.

models:
  - name: revenue_by_region
    access: public
    config:
      contract: {enforced: true}
    versions:
      - v: 1
        # the original columns
      - v: 2
        # v2 adds a currency column and drops the ordering guarantee
        columns:
          - name: region
            data_type: varchar
          - name: currency
            data_type: varchar
          - name: revenue
            data_type: number(38,2)
    latest_version: 2

Consumers pin the version they built against — ref('revenue_by_region', version=1) — and the two versions build as separate relations (revenue_by_region_v1, revenue_by_region_v2). You set a deprecation_date on v1, dbt warns anyone still referencing it, and when the last dashboard has moved you delete the v1 definition. That’s the whole point of the governance layer: you can change one part on a Friday, ship it as a new version behind a contract, and let consumers migrate on their own clock instead of yours.

Observability: knowing before the stakeholder does

Contracts stop you from breaking the shape. Observability catches the world breaking the data — a source that stopped loading, a join that fanned out, a metric that quietly went anomalous. The stack gives you four layers, and the craft is wiring them so the important ones page and the noisy ones don’t.

Freshness catches the source that stopped arriving. Half of all “the dashboard is wrong” incidents are really “the data is stale” — the transform ran fine, on yesterday’s data, because ingest silently died. dbt checks this directly:

# models/staging/_tpch__sources.yml
sources:
  - name: tpch
    database: SNOWFLAKE_SAMPLE_DATA
    schema: TPCH_SF1
    loaded_at_field: _loaded_at
    freshness:
      warn_after: {count: 12, period: hour}
      error_after: {count: 24, period: hour}
    tables:
      - name: orders

dbt source freshness compares max(_loaded_at) to now and warns at twelve hours, errors at twenty-four. Run it as its own step before the transform — a source that’s already a day stale is a run you’d rather skip than propagate. (The static SNOWFLAKE_SAMPLE_DATA has no load timestamp, so freshness is something you turn on once you’re transforming a real ingested source, which post 8 built.)

Tests carry a severity, and severity is the whole trick to alert sanity. A dbt test failing is not the same class of event as a task crashing, and treating them identically is how teams train themselves to ignore the pager. dbt lets each test declare how loudly it fails:

models:
  - name: orders_enriched
    columns:
      - name: order_key
        tests:
          - unique
          - not_null
      - name: total_price
        tests:
          - dbt_utils.accepted_range:
              min_value: 0
              config:
                severity: warn                 # negative price: suspicious, not fatal
                store_failures: true
      - name: customer_key
        tests:
          - relationships:
              to: ref('stg_customer')
              field: customer_id
              config:
                severity: error                # orphaned FK: the mart is wrong, stop

severity: error fails the build; severity: warn records the failure and keeps going. You can even make severity conditional — error_if: ">1000", warn_if: ">0" — so one bad row warns but a thousand pages. This is the distinction that keeps observability trustworthy: a broken foreign key means the marts are built on a lie and the run should go red; a handful of oddly-priced line items is worth a look, not a 2am wake-up.

store_failures turns tests into a queryable audit trail. By default a failed test is a red mark and a count in the logs. store_failures: true writes the offending rows to a table (in a dbt_test__audit schema by default), so instead of “12 rows failed accepted_range” you have the twelve rows. Build a thin monitoring model on top of those audit tables and you have a dashboard of your own data quality over time:

-- models/monitoring/test_failures.sql
{{ config(materialized='view') }}

select
    'orders_enriched.total_price_range' as test_name,
    count(*)                            as failing_rows,
    current_timestamp()                 as checked_at
from {{ ref('accepted_range_orders_enriched_total_price') }}
-- union additional stored-failure tables here

Now “is data quality getting worse” is a query, not a vibe.

Elementary adds the tests you didn’t know to write. Schema tests catch violations of rules you stated. Anomalies are the failures you didn’t think to assert — row volume that halved overnight, a freshness gap, a column whose null rate tripled. Elementary is a dbt package that installs anomaly-detection tests and, through an on-run-end hook, logs every test and run result into an elementary schema:

# packages.yml
packages:
  - package: elementary-data/elementary
    version: [">=0.16.0", "<0.17.0"]
# an anomaly test on the mart's daily volume
models:
  - name: revenue_by_region
    tests:
      - elementary.volume_anomalies:
          timestamp_column: checked_at

edr report renders an HTML observability report from that schema — test results, model run times, and anomaly flags — and edr monitor can push alerts to Slack. It’s the layer that watches the metrics you never wrote a rule for, which in practice is where the surprising incidents live: the day a source doubled its row count because of a bad upstream backfill, the morning a column’s null rate spiked because a producer renamed a field. No schema test you’d have thought to write catches those; a baseline the tool learned from history does. Because Elementary rides your existing dbt run — same warehouse, same on-run-end hook — it costs one extra pass over the audit schema, not a second monitoring system to operate.

Reconciling Airflow alerts with dbt test results. Here’s the seam that ties observability back to the orchestrator, and the good news is that it mostly works correctly already — for a reason worth understanding, because people routinely reach for the wrong knob here.

Cosmos renders each dbt test as its own Airflow task, and a failed task fires on_failure_callback. So the worry is obvious: will a severity: warn test — the kind that shouldn’t wake anyone — page like a crashed warehouse?

No, and severity is dbt’s business, not Cosmos’s. A test configured severity: warn does not fail dbt test or dbt build; dbt reports it as a warning and still exits zero. The Cosmos task wrapping it therefore stays green, and no failure callback fires. There is no Cosmos setting involved in that, and in particular TestBehavior is not the lever — it only decides how tests are rendered into tasks (a dbt build per node, a test task after each model, one at the end, or none at all). Setting TestBehavior.BUILD in the belief that it makes warnings non-paging is cargo-cult config: the behavior you wanted was already there.

The real problem is the opposite one. A warned test leaves the task green, which means nobody ever sees it. It scrolls past in a task log that only gets opened when something is red. If a warning is worth configuring, it’s worth surfacing, and Cosmos gives you a hook for exactly that:

from cosmos import DbtDag, RenderConfig

def warn_to_slack(context, test_names, test_results):
    send_slack(f"dbt warnings in {context['dag'].dag_id}: "
               + ", ".join(test_names))

dag = DbtDag(
    render_config=RenderConfig(),
    operator_args={"on_warning_callback": warn_to_slack},
    ...
)

on_warning_callback fires when a test warns but does not fail, and it receives the names and results of the warning tests. Route it somewhere low-noise — a #data-quality channel, the Elementary report, a monitoring model fed by store_failures — deliberately not the pager.

The rule to hold onto: infrastructure failures and data-quality errors page; data-quality warnings get logged and reviewed. dbt’s severity already implements that split for you; your job is to make sure the quiet half is still visible somewhere. Wire them the same and you’ll soon be muting the channel, which is worse than no alerting at all.

Deadline Alerts catch the run that never finished. Every alert above fires when something runs and fails. The nastier failure is the run that hangs, or never starts — a warehouse that won’t resume, an upstream asset that never arrives, so the pipeline is simply late with no error to catch. Airflow 2.x had SLAs for this; Airflow 3.x removed SLAs and replaced them with Deadline Alerts, which fix the old feature’s worst flaw — an SLA only evaluated after a task ran, so a task that never started never missed its SLA. A deadline is absolute:

from datetime import timedelta
from airflow.sdk import DAG
from airflow.sdk import DeadlineAlert, DeadlineReference

dag = DAG(
    dag_id="tpch_pipeline",
    schedule=[raw_orders],
    deadline=DeadlineAlert(
        reference=DeadlineReference.DAGRUN_LOGICAL_DATE,
        interval=timedelta(hours=2),
        callback=alert_pipeline_late,   # paged if not done 2h after the run's logical time
    ),
)

If the run hasn’t completed two hours after its logical date, alert_pipeline_late fires — whether the pipeline failed, stalled, or never woke up. That closes the last observability gap: you now hear about the run that errored (failure callbacks), the data that’s wrong (dbt tests), the data that’s anomalous (Elementary), the source that’s stale (freshness), and the run that’s simply not there (deadlines). (Deadline Alerts arrived in the Airflow 3.1 line and evolved across 3.x; the class path and reference options are worth checking against your exact 3.3 build, since the API firmed up release to release.)

Final thoughts

Seven series brought you here, and this last post is the one that makes the other six safe to touch. Look at what each layer actually bought: lineage gave you a graph that spans all three tools, so impact analysis is a query instead of a guess. Contracts gave the public marts a shape you can’t accidentally break and a versioning scheme for when you must break them on purpose. Observability gave you five different ways to learn a run went wrong, sorted so the urgent ones page and the routine ones don’t. Governed, versioned, and observable — that’s the difference between a pipeline that demos and one a business runs its numbers on.

Notice, one final time, that none of this lives in a tool. dbt emits the lineage and enforces the contracts; Airflow and Cosmos emit the run graph and route the alerts; Snowflake records who read what and runs every test as SQL; OpenLineage and Elementary and Marquez are the connective tissue that lets the three see each other. The platform was never any one of them. It’s the chain of evidence they form together — from raw arrival, through a modeled and contracted table, to the dashboard someone trusts — with enough seams in the right places that you can change one link on a Friday and know, before you merge, exactly what you touched and exactly who to tell.

That’s the whole stack. Snowflake for compute, dbt for transformation, Airflow and Cosmos for orchestration, and this last layer for the trust that lets you keep changing it. Build the seams deliberately and you don’t get a product a vendor sold you — you get a data platform you understand, one config file at a time. Go ship something.

Comments