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.

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:

The package also adds the following 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.1.0

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:

• 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.DbtCloudRunJobOperatorExtractorWith 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.

Step 3: Configure the dbt Cloud connection

Create or update the Airflow connection used by your DAG:

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:

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

Built-in dbt artifact extraction (optional)

Starting in acceldata-openlineage 1.1.0, the setting dbt_cloud_lineage_extraction_enabled controls whether the extractor delegates to Airflow's built-in dbt Cloud OpenLineage methods (get_openlineage_facets_on_start, get_openlineage_facets_on_complete, get_openlineage_facets_on_failure).

By default, built-in extraction is enabled (true). No change is required for existing deployments.

Set dbt_cloud_lineage_extraction_enabled to false when the Acceldata dbt Cloud Data Source connector already collects model, test, and dataset lineage in ADOC and you only need orchestration lineage plus the dbtCloud linking facet from Airflow. This avoids duplicate or conflicting dbt asset lineage from both Airflow and the Data Source connector.

Self-managed Airflow

Environment variable:

export AIRFLOW__OPENLINEAGE__DBT_CLOUD_LINEAGE_EXTRACTION_ENABLED=false

airflow.cfg

[openlineage]
dbt_cloud_lineage_extraction_enabled = false

Restart the scheduler and workers after changing this setting.

Google Cloud Composer

Set AIRFLOW__OPENLINEAGE__DBT_CLOUD_LINEAGE_EXTRACTION_ENABLED=false under Environment variables, then apply the environment update. Restart the scheduler and workers after changing this setting.

Step 4: Verify the installation

Verify the extractor import

from acceldata_openlineage import DbtCloudRunJobOperatorExtractorprint(DbtCloudRunJobOperatorExtractor.get_operator_classnames())
Expected output:
['DbtCloudRunJobOperator']

Run a DAG

Execute a DAG containing a DbtCloudRunJobOperator and confirm the following:

Task logs should include a message similar to:

Attached dbtCloud facet

Validate in ADOC

After the DAG completes, verify the following:

Compatibility

Troubleshooting

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.