Link Airflow Tasks to dbt Cloud Pipeline Runs in ADOC

When an Airflow task uses DbtCloudRunJobOperator with both the dbt Cloud data source and the Airflow OpenLineage integration configured, ADOC discovers both systems but cannot reliably associate the Airflow task with the corresponding dbt Cloud pipeline run. Without this association, the Airflow task and the dbt Cloud pipeline appear as separate lineage entities rather than a single end-to-end execution path.

This guide explains how to configure acceldata-openlineage, a lightweight OpenLineage extractor, so ADOC can associate Airflow tasks that use DbtCloudRunJobOperator with their corresponding dbt Cloud pipeline runs.

Why this happens

The current OpenLineage implementation for DbtCloudRunJobOperator emits lineage for dbt assets — including models, tests, snapshots, and sources — but does not emit a job-level event representing the dbt Cloud job itself.

Component	Behavior
Airflow OpenLineage	Emits DAG and task lineage
dbt Cloud data source	Discovers dbt Cloud jobs and runs
ADOC	Cannot deterministically associate an Airflow task with its corresponding dbt Cloud pipeline run

What the extractor does

acceldata-openlineage provides a custom OpenLineage extractor for DbtCloudRunJobOperator. The extractor:

Preserves the existing OpenLineage behavior.
Preserves dbt Cloud asset lineage.
Adds an Acceldata-specific dbtCloud run facet.
Enables ADOC to associate Airflow tasks with dbt Cloud pipeline runs.
Prevents duplicate pipeline entities.

Prerequisites

OpenLineage must already be configured in your Airflow environment. The following components are required:

Component	Requirement
Apache Airflow	2.10 or later
apache-airflow-providers-openlineage	Required for custom extractors
apache-airflow-providers-dbt-cloud	Required for `DbtCloudRunJobOperator`
acceldata-openlineage	The custom extractor package

The package also adds the following runtime dependency:

Package	Purpose
attrs	Runtime dependency

Step 1: Install the package

Install the package on every Airflow component that executes or schedules tasks: the scheduler, workers, and the triggerer, if used.

For production deployments, install a pinned release version:

pip install acceldata-openlineage==1.0.0

Install the package in the same Python environment used by Airflow, then restart the affected Airflow services.

Step 2: Register the extractor

Configure the OpenLineage provider to use the custom extractor. The extractor class is:

acceldata_openlineage.extractors.dbt.dbt_cloud.DbtCloudRunJobOperatorExtractor

If other extractors are already configured, append this class to the existing list using semicolons.

Self-managed Airflow

Environment variable:

export AIRFLOW__OPENLINEAGE__EXTRACTORS=acceldata_openlineage.extractors.dbt.dbt_cloud.DbtCloudRunJobOperatorExtractor

With additional extractors:

export AIRFLOW__OPENLINEAGE__EXTRACTORS=your.existing.Extractor;acceldata_openlineage.extractors.dbt.dbt_cloud.DbtCloudRunJobOperatorExtractor

airflow.cfg:

[openlineage] extractors = acceldata_openlineage.extractors.dbt.dbt_cloud.DbtCloudRunJobOperatorExtractor

Restart the scheduler and workers after updating the configuration.

Google Cloud Composer

Open Composer → Environments → your environment → Edit.
Add PyPI packages. Add the following packages:
Package
Notes
acceldata-openlineage
Pin a released version
apache-airflow-providers-dbt-cloud
Add it if not already installed
Composer installs these packages on all Airflow components during the environment update.
Configure the extractor. Set AIRFLOW__OPENLINEAGE__EXTRACTORS (or the equivalent alias, OPENLINEAGE_EXTRACTORS) to:
acceldata_openlineage.extractors.dbt.dbt_cloud.DbtCloudRunJobOperatorExtractor
With additional extractors:
your.existing.Extractor;acceldata_openlineage.extractors.dbt.dbt_cloud.DbtCloudRunJobOperatorExtractor
Apply the update. Save the Composer environment. Composer rebuilds the scheduler and workers, which may take several minutes.

Package	Notes
acceldata-openlineage	Pin a released version
apache-airflow-providers-dbt-cloud	Add it if not already installed

Step 3: Configure the dbt Cloud connection

Create or update the Airflow connection used by your DAG:

Field	Value
Connection ID	Match `dbt_cloud_conn_id`; the default is `dbt_cloud_default`
Connection type	dbt Cloud
Login	dbt Cloud Account ID
Password	dbt Cloud API token

If account_id is not explicitly set on the operator, the extractor uses the Account ID in the connection's Login field.

What the extractor adds

The extractor delegates to the built-in OpenLineage implementation, preserving existing lineage behavior. It also attaches a custom run facet named dbtCloud containing:

Field	Source
dbtCloudRunId	`operator.run_id`
dbtCloudJobId	`operator.job_id`
dbtCloudAccountId	`operator.account_id`, or the connection's Login field

This metadata enables ADOC to associate Airflow tasks with their corresponding dbt Cloud pipeline runs.

Step 4: Verify the installation

Verify the extractor import

from acceldata_openlineage import DbtCloudRunJobOperatorExtractor print(DbtCloudRunJobOperatorExtractor.get_operator_classnames())

Expected output:

['DbtCloudRunJobOperator']

Run a DAG

Execute a DAG containing a DbtCloudRunJobOperator and confirm the following:

Check	Expected result
Extractor loaded	No import errors appear in the Airflow logs
OpenLineage events	Events are emitted successfully
dbtCloud facet	The facet is present in emitted run events
wait_for_termination	True

Task logs should include a message similar to:

Attached dbtCloud facet

Validate in ADOC

After the DAG completes, verify the following:

Validation	Expected result
Task-to-pipeline linking	The Airflow task links to the corresponding dbt Cloud pipeline run
Pipeline lineage	Existing dbt Cloud pipeline lineage is preserved
Asset lineage	Models, tests, snapshots, and sources continue to appear normally
Duplicate pipelines	No duplicate dbt Cloud pipeline entities are created

Compatibility

Item	Details
Airflow	Compatible with Airflow 2.10.x
dbt Cloud provider	No provider upgrade is required
ADOC dbt Cloud data source	Fully supported

Troubleshooting

Symptom	Resolution
Extractor is not loading	Verify the configured class name matches exactly, and that the package is installed on all Airflow workers
No dbtCloud facet on the START event	This is expected when `run_id` is not yet available. The facet is typically attached to COMPLETE or FAIL events
Missing dbtCloudAccountId	Set `account_id` on the operator, or populate the Airflow connection's Login field with the dbt Cloud Account ID
Composer changes are not applied	Confirm the Composer environment update completed successfully
No task-to-pipeline linking in ADOC	Verify that OpenLineage events include the dbtCloud facet, and that the dbt Cloud data source is configured

Acceptance criteria

The installation is successful when:

Airflow tasks are linked to their corresponding dbt Cloud pipeline runs in ADOC.
Existing dbt Cloud pipeline lineage is preserved.
Existing dbt asset lineage remains unchanged.
No duplicate dbt Cloud pipeline entities are created.
The solution works without requiring an Airflow or dbt Cloud provider upgrade.

Sample DAG

from datetime import datetime from airflow import DAG from airflow.providers.dbt.cloud.operators.dbt import DbtCloudRunJobOperator with DAG( dag_id='dbt_cloud_dag', start_date=datetime(2026, 7, 6), schedule=None, catchup=False, ) as dag: dbt_cloud_job = DbtCloudRunJobOperator( task_id='dbt_cloud_job_run', job_id=70471823610018, # Replace with your dbt Cloud job ID dbt_cloud_conn_id='dbt_cloud_default', wait_for_termination=True, check_interval=10, timeout=300, )

Once configured, the Airflow task and its corresponding dbt Cloud pipeline run appear linked in the ADOC pipeline view, as shown below.

What's next

For general Airflow OpenLineage setup, see Airflow OpenLineage Integration.
For dbt Core (rather than dbt Cloud) OpenLineage setup, see dbt Core OpenLineage Integrations.

Last updated on

Was this page helpful?