Acceldata OpenLineage

Custom OpenLineage extractors for Acceldata. Install this package in your Airflow environment and register the extractors so lineage events include Acceldata-specific facets for dbt Cloud runs.

Prerequisites

OpenLineage should already be enabled and configured in your Airflow environment. This package additionally requires:

Component	Notes
Apache Airflow	2.10 or later
apache-airflow-providers-openlineage	Required for custom extractors
apache-airflow-providers-dbt-cloud	Required for DbtCloudRunJobOperator

This package adds one runtime dependency: attrs.

Install

Install on every Airflow component that runs or schedules tasks (workers, scheduler, and triggerer if applicable):

pip install acceldata-openlineage

Self-managed Airflow: install into the same Python environment used by Airflow processes, then restart workers and the scheduler.

For Google Cloud Composer, follow the dedicated section below.

Setup on Google Cloud Composer

These steps assume OpenLineage is already configured in your Composer environment. Changes are made in the Google Cloud Console under Composer → Environments → your environment → Edit.

1. Add PyPI packages

Open the PyPI packages tab and add:

Package	Notes
acceldata-openlineage	This package (pin a version in production, e.g. ==1.0.0)
apache-airflow-providers-dbt-cloud	Required if not already installed

Composer installs packages on all Airflow components during the next environment update.

2. Register the extractor

Open the Environment variables tab and set OPENLINEAGE_EXTRACTORS (AIRFLOW__OPENLINEAGE__EXTRACTORS is an equivalent alias).

Append this extractor to any extractors you already use—do not replace the full list unless you intend to:

acceldata_openlineage.extractors.dbt.dbt_cloud.DbtCloudRunJobOperatorExtractor

Example when other extractors are already registered:

your.existing.Extractor;acceldata_openlineage.extractors.dbt.dbt_cloud.DbtCloudRunJobOperatorExtractor

3. Configure the dbt Cloud connection

In the Airflow UI (Admin → Connections), create or update the connection used by your DAGs:

Field	Value
Connection Id	Same as dbt_cloud_conn_id on the operator (default: dbt_cloud_default)
Connection Type	dbt Cloud
Login	dbt Cloud Account ID
Password	dbt Cloud API token

When account_id is not set on the operator, the extractor resolves dbtCloudAccountId from the connection login.

4. Apply the environment update

Save the Composer environment changes. Composer rebuilds workers and the scheduler; this can take several minutes.

After the update completes, run a DAG with a DbtCloudRunJobOperator task and check task logs for Attached dbtCloud facet.

Register extractors (other environments)

For self-managed Airflow, MWAA, Astronomer, or other platforms, register the extractor via environment variable or airflow.cfg.

Set a semicolon-separated list of fully qualified class names. Append this extractor to any extractors you already use:

Environment variable:

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

If you already have extractors configured, append with ;:

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

OPENLINEAGE_EXTRACTORS is an equivalent alias supported by the OpenLineage provider.

Restart Airflow workers and the scheduler after changing extractor configuration.

dbt Cloud connection

Configure a dbt Cloud Airflow connection as usual:

Connection field	Value
Conn Id	Same as dbt_cloud_conn_id on the operator (default: dbt_cloud_default)
Login	dbt Cloud Account ID
Password	dbt Cloud API token

When account_id is not set on the operator, the extractor resolves dbtCloudAccountId from the connection login (same behavior as the dbt Cloud provider hook).

Included extractors

Extractor	Operator	Facet key
DbtCloudRunJobOperatorExtractor	DbtCloudRunJobOperator	dbtCloud

dbtCloud run facet

Attached when operator.run_id is set (normally on COMPLETE / FAIL; also on START if the run was reused). Fields:

Field	Source
dbtCloudRunId	DbtCloudRunJobOperator.run_id
dbtCloudJobId	DbtCloudRunJobOperator.job_id
dbtCloudAccountId	DbtCloudRunJobOperator.account_id, or the Account ID (login) from dbt_cloud_conn_id when unset

The extractor delegates to the operator's built-in OpenLineage methods so default dbt Cloud lineage behavior is preserved, then merges the Acceldata facet used for InterPipelineRunLink.

Verify installation

On an Airflow worker node, confirm the package and extractor import correctly:

from acceldata_openlineage import DbtCloudRunJobOperatorExtractor

print(DbtCloudRunJobOperatorExtractor.get_operator_classnames())
# ['DbtCloudRunJobOperator']

Run a DAG with a DbtCloudRunJobOperator task and confirm lineage events include a dbtCloud run facet with dbtCloudRunId, dbtCloudJobId, and dbtCloudAccountId populated.

Troubleshooting

Symptom	What to check
Extractor not loading	Import path matches config exactly; package installed on the worker running the task
No dbtCloud facet on START	Expected when run_id is not set yet; facet appears on COMPLETE / FAIL
Missing dbtCloudAccountId	Set account_id on the operator, or put the Account ID in the connection Login field
Composer: changes not picked up	Environment update finished successfully

Check Airflow task logs for messages such as Attached dbtCloud facet or warnings about connection lookup failures.