acceldata-openlineage

Acceldata OpenLineage

Custom OpenLineage extractors for Acceldata. Adds a dbtCloud run facet to DbtCloudRunJobOperator lineage events so ADOC can link Airflow tasks to dbt Cloud runs (InterPipelineRunLink).

By default, the extractor also delegates to Airflow's built-in dbt Cloud OpenLineage integration. If the Acceldata dbt Cloud Data Source connector already collects detailed dbt lineage, you can disable that delegation (see below).

Prerequisites

OpenLineage must already be enabled in Airflow, plus:

Install

Install on workers, scheduler, and triggerer (if used):

pip install acceldata-openlineage==1.1.0

Restart Airflow processes after install.

Setup

1. Register the extractor

Append to existing extractors (semicolon-separated); do not replace the full list unless intended:

# airflow.cfg
[openlineage]
extractors = acceldata_openlineage.extractors.dbt.dbt_cloud.DbtCloudRunJobOperatorExtractor

Or via environment variable:

AIRFLOW__OPENLINEAGE__EXTRACTORS=acceldata_openlineage.extractors.dbt.dbt_cloud.DbtCloudRunJobOperatorExtractor

Composer: add the package under PyPI packages, set AIRFLOW__OPENLINEAGE__EXTRACTORS under Environment variables, then apply the environment update.

2. dbt Cloud connection

If account_id is unset on the operator, dbtCloudAccountId is taken from the connection login.

3. Verify

Run a DbtCloudRunJobOperator task and check logs for Attached dbtCloud facet. On COMPLETE / FAIL the facet should include dbtCloudRunId, dbtCloudJobId, and dbtCloudAccountId.

Built-in dbt artifact extraction (optional)

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_*).

Set to false when the Acceldata dbt Cloud Data Source connector already collects model/test/dataset lineage in ADOC and you only need orchestration lineage plus the dbtCloud linking facet from Airflow.

[openlineage]
dbt_cloud_lineage_extraction_enabled = false

AIRFLOW__OPENLINEAGE__DBT_CLOUD_LINEAGE_EXTRACTION_ENABLED=false

Restart scheduler and workers after changing this.

Note: built-in extraction also requires wait_for_termination=True on the operator. With wait_for_termination=False, artifact extraction is already a no-op regardless of this flag.

dbtCloud facet

Present on COMPLETE / FAIL once run_id is set (START only if the run was reused).

Troubleshooting

• No facet — confirm the extractor path in config, package on the worker, and task reached COMPLETE / FAIL with a run_id.
• Missing dbtCloudAccountId — set account_id on the operator or Account ID in the connection login.
• Flag has no effect — set wait_for_termination=True to test; with False, built-in extraction never runs anyway.
• Duplicate lineage or upstream errors (KeyError: 'host', catalog.json 404) — set dbt_cloud_lineage_extraction_enabled = false.