Assets, Lineage, and the Cross-Tool Seam

How Cosmos emits Airflow assets for dbt models, how data-aware scheduling bridges tools, and where OpenLineage fits.

Every DAG so far has run on a clock. @daily, 0 2 * * *, a cron string that says when — and hopes the upstream data is ready by then. That hope is the weakest joint in the whole pipeline. The nightly Fivetran sync usually finishes by 1:30, so you schedule dbt for 2:00 with a comfortable margin, and then one morning the sync runs long, dbt fires at 2:00 anyway, and your orders mart rebuilds on yesterday’s rows. Nobody wrote a bug. The clock just guessed wrong.

The fix is to stop scheduling on time and start scheduling on data. When the orders model finishes and its table is genuinely updated, the reverse-ETL sync that pushes marts back to the CRM should run — not at a time you picked, but at the moment the thing it depends on became ready. Airflow 3.x calls the unit of data an Asset, and Cosmos hands you one per dbt model for free. This chapter is about that seam: how Cosmos emits assets, how a separate DAG downstream schedules on them, and how OpenLineage lets you trace a value from the loader that ingested it all the way to the dashboard that reads it.

Cosmos emits an asset per model

You already have assets and didn’t know it. RenderConfig carries a parameter, emit_datasets, and it defaults to True. (The name is a holdover — Airflow renamed Datasets to Assets in 3.0, but Cosmos kept the parameter spelling for back-compat. What it emits in an Airflow 3.x deployment is an Asset.) With it on, every model task Cosmos renders declares the model’s table as an outlet: when the task succeeds, Airflow records that the corresponding asset was updated.

The asset needs a stable identity, and this is the part worth getting exactly right, because a URI that is almost correct produces a consumer DAG that simply never fires — no error, no warning, just a schedule that waits forever.

Cosmos does not invent its own naming scheme. It derives the URI from the OpenLineage event dbt emits for the model, which means the identity is the physical table, not the dbt project. The namespace identifies the warehouse; the name identifies the relation:

<openlineage-namespace>/<database>/<schema>/<model>

For a Postgres target the namespace is postgres://<host>:<port>, so the bookshop’s orders mart lands as:

postgres://bookshop-db.internal:5432/bookshop/analytics/orders

Two details will bite you. First, that’s the Airflow 3 form: OpenLineage names the relation bookshop.analytics.orders, with dots, and Airflow 3 tightened its URI validation rules, so Cosmos rewrites the dots to slashes. Under Airflow 2 the same asset was …:5432/bookshop.analytics.orders. If you are porting a 2.x DAG, every hand-written asset URI in it is now wrong. (Cosmos is loud about this: it logs a warning at task run naming the old and new URIs.)

Second — and this is the practical advice — don’t hand-write it at all. It is derived from the host, port, database, schema, and model, any of which can differ between your dev and prod connections. Run the producer DAG once and read the exact URI off the task log or the Assets page in the UI, then paste that into your consumer’s schedule=. Typing it from memory is the single most common way this feature appears broken.

Nothing about the URI is magic once you have it — it’s a string, and its only job is to be the same string on both sides of the seam. The producer task attaches it as an outlet; the consumer DAG names it as a schedule. Airflow matches them by exact URI. That’s the whole contract, and the reason it’s worth a paragraph is that “exact” really does mean exact.

You don’t write any of this by hand for the producer. It’s the ordinary DbtDag from earlier chapters, with emit_datasets left at its default:

from pathlib import Path

from cosmos import DbtDag, ProjectConfig, ProfileConfig, RenderConfig
from cosmos.profiles import PostgresUserPasswordProfileMapping

DBT_PROJECT_PATH = Path("/usr/local/airflow/include/dbt/bookshop")

profile_config = ProfileConfig(
    profile_name="bookshop",
    target_name="dev",
    profile_mapping=PostgresUserPasswordProfileMapping(
        conn_id="bookshop_pg",
        profile_args={"schema": "analytics"},
    ),
)

bookshop_dbt = DbtDag(
    dag_id="bookshop_dbt",
    project_config=ProjectConfig(DBT_PROJECT_PATH),
    profile_config=profile_config,
    render_config=RenderConfig(emit_datasets=True),  # the default, shown for clarity
    schedule="@daily",
    start_date=None,
)

Open the grid after a run and the orders task now shows an asset in its outlets — Airflow’s UI has an Assets view that lists every declared asset and, for each, the tasks that produce it and the DAGs that consume it. Before you write a single consumer, you can already see the bookshop graph re-expressed as data: five models, five assets, each one a node another DAG can subscribe to.

One prerequisite, easy to miss. Everything in this chapter requires ExecutionMode.LOCAL or VIRTUALENV. Asset emission is implemented on Cosmos’s local operator base class only — the Kubernetes and Docker operators don’t inherit it, so if chapter 08 talked you into containerising dbt, the assets silently stop being emitted and every consumer below stops firing. Pick your execution mode with this chapter in mind, not after it.

The consumer schedules on the asset, not the clock

Here’s the move that makes this worth the chapter. A separate DAG — different file, different schedule, possibly a different team’s — reverse-ETLs the orders mart into the CRM so sales sees fresh order totals. It has no business running on a clock. It should run exactly when orders is rebuilt, and never otherwise. So instead of a cron string, its schedule is a list of assets:

from airflow.sdk import Asset, dag, task

ORDERS = Asset("postgres://bookshop-db.internal:5432/bookshop/analytics/orders")


@dag(schedule=[ORDERS], start_date=None, catchup=False)
def crm_order_sync():
    @task
    def push_to_crm():
        # read the freshly-built orders mart, upsert into the CRM
        ...

    push_to_crm()


crm_order_sync()

That schedule=[ORDERS] is the entire mechanism. This DAG has no @daily, no cron, no start-time margin. Airflow watches that orders asset, and the instant the producing task in bookshop_dbt succeeds and marks that asset updated, the scheduler creates a run of crm_order_sync. If the dbt DAG runs late, the sync runs late with it. If dbt is skipped because a freshness gate stopped it, the sync doesn’t run at all — there was no new data to push. The dependency is expressed as data readiness, and the guessing is gone.

Notice what this is not. It’s not a TriggerDagRunOperator bolted onto the end of the dbt DAG, and it’s not an ExternalTaskSensor in the sync DAG polling the scheduler and asking “is dbt done yet?” Both of those couple the two DAGs directly — the producer has to know its consumers, or the consumer has to know the producer’s task ids and schedule. Asset scheduling inverts that. bookshop_dbt doesn’t know crm_order_sync exists; it just declares “I update orders.” crm_order_sync doesn’t know how orders gets built; it just declares “I run when orders updates.” The asset URI is the only thing they share, and either side can change freely as long as that string holds. This is the difference between wiring two DAGs together and letting them meet at a named piece of data.

Waiting on several models at once

One asset is the easy case. The real seam usually needs more than one. A BI refresh that rebuilds a dashboard reads both orders and customers, and it should fire only when both have been rebuilt — running it after orders alone would refresh half the dashboard against stale customer data. Airflow expresses that with conditional asset expressions. Pass a list and the default is “any of these” — the DAG runs when any asset in the list updates. For “all of these,” combine them:

from airflow.sdk import Asset, AssetAll, dag, task

ORDERS = Asset("postgres://bookshop-db.internal:5432/bookshop/analytics/orders")
CUSTOMERS = Asset("postgres://bookshop-db.internal:5432/bookshop/analytics/customers")


@dag(schedule=AssetAll(ORDERS, CUSTOMERS), start_date=None, catchup=False)
def bi_dashboard_refresh():
    @task
    def refresh_dashboard():
        ...

    refresh_dashboard()


bi_dashboard_refresh()

AssetAll(ORDERS, CUSTOMERS) means the run waits until both assets have updated since the last time it fired — the BI refresh happens once, after the slower of the two marts lands, not twice. Its sibling AssetAny is the explicit form of the default: fire when either updates, which is what you want for a cache-invalidation job that should react to any change in a set of tables. The two compose, so you can express “when orders updates, and also when either customers or promotions updates” as AssetAll(ORDERS, AssetAny(CUSTOMERS, PROMOTIONS)). You’re describing the data precondition for the job in the same language dbt uses to describe its graph — declaratively, by naming the tables, not by encoding a schedule that approximates when they’ll be ready.

The phrase “since the last time it fired” hides the one bit of AssetAll semantics worth internalizing, because it’s where people trip. The condition tracks updates relative to the consumer’s last run. After the BI refresh fires, both assets’ counters reset for that DAG; it won’t fire again until it has seen a fresh update to both orders and customers. So if orders rebuilds three times before customers rebuilds once, the refresh still runs exactly once — the extra orders updates don’t stockpile into multiple runs, they just keep the “orders is ready” flag set until customers finally lands and completes the pair. That’s usually what you want: the refresh reflects the latest state of both tables, not a run per upstream event. But it means AssetAll is not a counter of events; it’s a gate on readiness. If you genuinely need one downstream run per upstream build, you don’t want AssetAll across several assets — you want the consumer to schedule on the single asset that paces the pipeline.

There’s a matching subtlety on the producer side worth stating plainly: Cosmos emits an asset for each model (and seed and snapshot), but not for the test tasks it renders alongside them. So under TestBehavior.AFTER_EACH, the model’s asset is marked updated when the model builds — the immediately-following test task guards the data, but it isn’t the thing that fires the asset. In practice the ordering saves you: the model task and its tests sit in the same DAG run, and a failing test upstream will fail the run before a downstream time-based dependency matters. But if you’re routing a reverse-ETL sync purely on a model’s asset, know that “asset updated” means “the model built,” not “the model built and passed its tests.” When that distinction matters — when you truly must not push untested data to the CRM — keep the sync in the same DAG as the dbt TaskGroup so Airflow’s ordinary task dependencies put it after the tests, and reserve asset scheduling for the looser, cross-DAG coupling where a build is a good-enough signal.

The whole seam, owned by Airflow

Step back and look at the shape this builds. A production analytics pipeline is almost never one tool. It’s a chain:

Fivetran / ingest  →  dbt (Cosmos)  →  reverse-ETL + BI refresh

The loader lands raw tables. dbt transforms them into marts. Downstream jobs push those marts back into operational systems and refresh dashboards. Three tools, three teams’ worth of concerns, and historically three schedules that each guessed at the previous one’s finish time — Fivetran at midnight, dbt at 2:00 “to be safe,” the BI refresh at 4:00 for the same reason. Two hours of padding, and it still breaks the night the sync runs long.

Assets let Airflow own the entire seam without a single time-based guess. The ingest side is the one link that often lives outside Airflow, and there’s a clean way to bring it in: have the ingestion — a Fivetran-triggering DAG, or an @task that kicks the connector and waits — declare the raw table as an asset outlet:

from airflow.sdk import Asset, dag, task

RAW_ORDERS = Asset("fivetran://bookshop/raw/orders")


@dag(schedule="0 0 * * *", start_date=None, catchup=False)
def ingest_orders():
    @task(outlets=[RAW_ORDERS])
    def sync_fivetran():
        # trigger the connector, poll to completion
        ...

    sync_fivetran()


ingest_orders()

Now the ingestion clock is the only clock in the chain, and it belongs where the actual external constraint lives — Fivetran’s sync window. Everything after it is data-driven. The dbt DAG schedules on Asset("fivetran://bookshop/raw/orders") instead of @daily; when it finishes it emits its per-model assets; the reverse-ETL and BI DAGs schedule on those. One time trigger at the head, and the readiness of data carries the pipeline the rest of the way. Airflow becomes the thing that knows the whole seam — not because it runs every tool, but because every tool’s output is a named asset it can route on.

That’s the conceptual payoff of Assets over the old sensor-and-trigger patterns: the pipeline’s structure is declared rather than wired. You can add a new consumer of orders — a data-science feature job, say — by writing one DAG with schedule=[ORDERS], and nothing upstream changes or even notices. The seam grows by addition, not by editing the producers.

OpenLineage: seeing across the tools you route between

Asset scheduling connects the tools. It doesn’t, on its own, let you see through them. When the CRM sync pushes wrong numbers, the question is “which upstream column fed this, and what touched it?” — and answering it means tracing a value across Fivetran, dbt, and the warehouse, three systems that each hold only their slice of the story. OpenLineage is the open standard for stitching those slices together, and both Airflow and Cosmos speak it.

Airflow’s side ships as a provider, apache-airflow-providers-openlineage. Install it and point it at a collector, either through airflow.cfg or environment variables:

[openlineage]
transport = {"type": "http", "url": "http://marquez:5000"}

With that in place, Airflow emits a lineage event for every task run — start, complete, fail — carrying the task’s inlets and outlets as OpenLineage datasets. Because Cosmos already declares each model’s table as an outlet, those events describe the dbt graph at the table level for free.

Cosmos adds the layer Airflow can’t see on its own: what happened inside dbt. Its operators read dbt’s run_results.json and the project manifest after each model builds and translate them into OpenLineage facets — which sources a model read, which model it wrote, the SQL that did it, and, where the warehouse and dbt metadata support it, column-level lineage. So the event for the orders task doesn’t just say “wrote analytics.orders”; it says “read int_order_payments and stg_customers, wrote orders, and here is the SQL and the columns that map through.” Airflow contributes the cross-DAG, cross-tool edges; Cosmos contributes the intra-dbt detail. Together they emit a connected graph.

You view that graph in an OpenLineage backend — Marquez is the reference implementation, a standalone UI that ingests OpenLineage events and draws the lineage. Open it and you get a dataset-level map that crosses tool boundaries: raw.orders (landed by Fivetran) → stg_orders → int_order_payments → orders → the CRM sync’s read. Click a dataset and you see its producers, its consumers, and its run history; where column lineage is present, you can follow a single field — orders.total_amount — back to the raw column it originated from.

It helps to know what’s actually crossing the wire, because “lineage” can sound like magic until you see it’s just structured events. Each OpenLineage event is a small JSON document naming a run, the job that produced it, and the input and output datasets, each of which carries facets — typed bits of metadata. A schema facet lists a dataset’s columns and types; a SQL facet carries the query that built it; a column-lineage facet maps each output column to the input columns it derived from. Cosmos populates these by parsing what dbt already writes to target/run_results.json and target/manifest.json after a build — it isn’t re-parsing your SQL, it’s reading dbt’s own record of what ran and what it touched, and reshaping it into the OpenLineage vocabulary. That’s why the integration is cheap and why it’s only as complete as dbt’s metadata: models dbt understands fully get column lineage; a model built through a macro or a warehouse feature dbt can’t introspect gets table-level lineage and no column detail. The graph is honest about what it knows, which is the property you want from a debugging tool.

Why this matters at the seam specifically: it turns two questions from archaeology into a click.

  • Impact analysis, forward. Someone proposes renaming a column in stg_orders. Lineage shows every dataset downstream — int_order_payments, orders, the CRM sync, the dashboard — before the change ships, so “what breaks if I touch this” has an answer that isn’t “let’s find out in production.”
  • Debugging, backward. The CRM shows a wrong order total. Lineage walks the chain in reverse: orders ← int_order_payments ← stg_orders ← raw.orders, with the run that produced each. You find the model that introduced the bad value instead of bisecting six tables by hand.

Airflow’s Assets view and OpenLineage answer related but distinct questions, and it’s worth keeping them straight. Assets are the scheduling graph — what triggers what, resolved by URI, used to route runs. OpenLineage is the observability graph — what derived from what, resolved down to columns, used to understand and debug. Cosmos feeds both from the same dbt project, which is exactly why a task-per-model rendering pays for its parse cost: one dbt graph, re-expressed as both a schedule and a lineage map.

When assets are the wrong tool

Asset scheduling is clean when Airflow can see both sides of the seam. It gets awkward — or plain wrong — when it can’t, and it’s worth naming the boundaries before you route your whole platform on it.

The first is producers Airflow doesn’t run. Asset scheduling works because a task with an outlet marks the asset updated on success. If the raw tables land via a Fivetran sync that Airflow neither triggers nor watches, there’s no task to attach an outlet to, and the asset never fires. You have two honest options: wrap the ingestion in an Airflow task that triggers and polls the connector (the outlets=[RAW_ORDERS] pattern above), or accept that the ingest→dbt link stays a clock or an external signal and let asset scheduling take over only after dbt, where Airflow is genuinely in control. What you shouldn’t do is pretend an asset represents freshness it can’t observe — an asset event means “an Airflow task said so,” not “the table is actually current.”

The second is cross-team, cross-deployment coupling. Asset scheduling lives inside one Airflow deployment; a consumer DAG and the producer whose asset it watches have to run under the same scheduler. When the data platform spans teams on separate Airflow instances — or crosses into tools with no Airflow at all — the shared URI stops being a shared reality, and you’re back to explicit contracts: an event on a message bus, an API call, a data contract with its own SLA. Assets are a within-Airflow coordination primitive, not a company-wide event bus, and stretching them across org boundaries builds a dependency that looks declarative but is actually a handshake nobody documented.

For those cases, Cosmos gives you the two escape hatches. Setting emit_datasets=False on RenderConfig turns off asset emission entirely — the right call when nothing schedules on dbt’s outputs and you’d rather not clutter the Assets view with URIs no DAG consumes:

render_config = RenderConfig(emit_datasets=False)

And when assets are consumed but the auto-generated {conn_id}://{project}/{model} URI doesn’t match what the rest of your platform expects — because a downstream tool keys on the warehouse’s fully-qualified table name, or because a naming convention predates Cosmos — you override the URI so both sides agree on the string. The identity of an asset is a contract, and the moment two systems depend on it, it deserves to be chosen deliberately rather than inherited from whatever Cosmos happened to generate. A URI that changes when you swap a connection id is a URI that will silently break a consumer the day you migrate warehouses.

Examples in this chapter are docs-checked against the series’ stated versions, but they were not executed in this repository unless a companion project explicitly says so.

Final thoughts

The clock was always a stand-in. 2 a.m. never meant “2 a.m.”; it meant “after the data is ready, and this is my best guess at when that is.” Assets let you delete the guess and say the true thing: this runs when this data is ready. Cosmos hands you an asset per model without asking, a downstream DAG schedules on it with a one-line schedule=[Asset(...)], and AssetAll/AssetAny let a job wait on exactly the set of models it reads. The ingest→dbt→consume seam that used to be three padded schedules praying at each other becomes one time trigger at the head and pure data-readiness after — a pipeline Airflow owns end to end without running every tool in it.

OpenLineage is the other half of the same story. Asset scheduling connects the tools; lineage lets you see through them, so a wrong number in the CRM traces back to the model that introduced it and a proposed column rename shows its blast radius before it ships. Cosmos feeds both graphs from the one dbt project — the scheduling graph and the observability graph, the same models re-expressed twice.

And the discipline underneath all of it is naming ownership. A good cross-tool seam says which DAG produces each asset, which dbt model defines it, which tests guard it, and which downstream consumers are allowed to trust it. Get those names right and the asset URI becomes a real contract — one that survives a warehouse migration, a team handoff, a new consumer nobody told you about. Get them wrong and you’ve built the same brittle coupling as before, only now it looks declarative, which is worse, because it fails quietly. The tool is good. The naming is the work.

Next: Incremental dbt Under Airflow Backfills

Comments