# Airflow Integration

:::note

If you're looking to schedule DataHub ingestion using Airflow, see the guide on [scheduling ingestion with Airflow](../../metadata-ingestion/schedule_docs/airflow.md).

:::

The DataHub Airflow plugin supports:

- Automatic column-level lineage extraction from various operators e.g. SQL operators (including `MySqlOperator`, `PostgresOperator`, `SnowflakeOperator`, `BigQueryInsertJobOperator`, and more), `S3FileTransformOperator`, and more.
- Airflow DAG and tasks, including properties, ownership, and tags.
- Task run information, including task successes and failures.
- Manual lineage annotations using `inlets` and `outlets` on Airflow operators.

The plugin requires Airflow 3.0+ and Python 3.10+. If you're on an older Airflow release, pin an older version of `acryl-datahub-airflow-plugin` — see the [compatibility section](#compatibility) for details.

<!-- TODO: Update the local Airflow guide and link to it here. -->
<!-- If you are looking to run Airflow and DataHub using docker locally, follow the guide [here](../../docker/airflow/local_airflow.md). -->

## DataHub Plugin Setup

### Installation

The plugin requires Airflow 3.0+ and Python 3.10+. If you don't meet these requirements, see the [compatibility section](#compatibility) for other options.

```shell
pip install 'acryl-datahub-airflow-plugin'
```

### Installing with Airflow constraints

`acryl-datahub` requires `click>=8.3.3` and `anyio>=4.14.2`. Apache Airflow's published constraint file still pins older releases of both, so `pip install acryl-datahub-airflow-plugin -c <airflow constraints>` cannot resolve. Rewrite those two lines, then install against the result. Replace the Airflow and Python versions in the URL with the ones you run:

```shell
curl -fsSL "https://raw.githubusercontent.com/apache/airflow/constraints-3.0.2/constraints-3.11.txt" \
  | sed -e 's/^anyio==.*/anyio>=4.14.2/' -e 's/^click==.*/click>=8.3.3/' \
  > /tmp/datahub-relaxed-constraints.txt
pip install 'acryl-datahub-airflow-plugin' -c /tmp/datahub-relaxed-constraints.txt
```

Leave every other pin in that file as Airflow published it.

### Configuration

Set up a DataHub connection in Airflow, either via command line or the Airflow UI.

#### Command Line

```shell
airflow connections add  --conn-type 'datahub-rest' 'datahub_rest_default' --conn-host 'http://datahub-gms:8080' --conn-password '<optional datahub auth token>'
```

If you are using DataHub Cloud then please use `https://YOUR_PREFIX.acryl.io/gms` as the `--conn-host` parameter.

#### Airflow UI

On the Airflow UI, go to Admin -> Connections and click the "+" symbol to create a new connection. Select "DataHub REST Server" from the dropdown for "Connection Type" and enter the appropriate values.

<p align="center">
  <img width="70%"  src="https://raw.githubusercontent.com/datahub-project/static-assets/main/imgs/airflow/plugin_connection_setup.png"/>
</p>

#### Optional Configurations

No additional configuration is required to use the plugin. However, there are some optional configuration parameters that can be set in the `airflow.cfg` file.

```ini title="airflow.cfg"
[datahub]
# Optional - additional config here.
enabled = True  # default
```

| Name                               | Default value        | Description                                                                                                                                                                                                                                               |
| ---------------------------------- | -------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| enabled                            | true                 | If the plugin should be enabled.                                                                                                                                                                                                                          |
| conn_id                            | datahub_rest_default | The name of the datahub rest connection.                                                                                                                                                                                                                  |
| cluster                            | prod                 | name of the airflow cluster, this is equivalent to the `env` of the instance                                                                                                                                                                              |
| platform_instance                  | None                 | The instance of the platform that all assets produced by this plugin belong to. It is optional.                                                                                                                                                           |
| capture_ownership_info             | true                 | Extract DAG ownership.                                                                                                                                                                                                                                    |
| capture_ownership_as_group         | false                | When extracting DAG ownership, treat DAG owner as a group rather than a user                                                                                                                                                                              |
| capture_tags_info                  | true                 | Extract DAG tags.                                                                                                                                                                                                                                         |
| capture_executions                 | true                 | Extract task runs and success/failure statuses. This will show up in DataHub "Runs" tab.                                                                                                                                                                  |
| materialize_iolets                 | true                 | Create the datasets referenced by task inlets/outlets if they are missing from DataHub. See [Materializing lineage iolets](#materializing-lineage-iolets).                                                                                                |
| enable_extractors                  | true                 | Enable automatic lineage extraction.                                                                                                                                                                                                                      |
| disable_openlineage_plugin         | true                 | Disable the OpenLineage plugin to avoid duplicative processing.                                                                                                                                                                                           |
| enable_multi_statement_sql_parsing | false                | Parse multiple SQL statements within a single task. Resolves temp tables and merges lineage across statements in one execution.                                                                                                                           |
| log_level                          | _no change_          | [debug] Set the log level for the plugin.                                                                                                                                                                                                                 |
| debug_emitter                      | false                | [debug] If true, the plugin will log the emitted events.                                                                                                                                                                                                  |
| dag_filter_str                     | { "allow": [".*"] }  | AllowDenyPattern value in form of JSON string to filter the DAGs from running.                                                                                                                                                                            |
| dataset_pattern                    | { "allow": [".*"] }  | AllowDenyPattern value in form of JSON string matched against `<platform>:<name>` of every inlet and outlet. See [Filtering inlets and outlets](#filtering-inlets-and-outlets).                                                                           |
| normalize_object_storage_urns      | true                 | Map OpenLineage S3, GCS, Azure and DBFS paths to the `s3`, `gcs`, `abs` and `dbfs` platforms and keep the bucket in the name, so they match datasets from DataHub's storage sources and the Spark agent. Set to `false` to keep the pre-1.8 URNs.         |
| capture_ol_file_datasets           | false                | Keep `file` datasets that operators report through OpenLineage. These are usually scratch paths on the worker, such as `/tmp/tmpab12/out.csv`, so they are skipped by default. Files declared as Airflow Assets or manual inlets/outlets are always kept. |
| capture_bigquery_temp_datasets     | false                | Keep BigQuery tables in hidden and anonymous datasets, which hold query results, scripts and sessions. They are dropped by default.                                                                                                                       |
| bigquery_temp_table_dataset_prefix | \_                   | Dataset-name prefix that marks those hidden datasets. Matches the BigQuery source's `temp_table_dataset_prefix`.                                                                                                                                          |
| enable_datajob_lineage             | true                 | If true, the plugin will emit input/output lineage for DataJobs.                                                                                                                                                                                          |
| capture_airflow_assets             | true                 | Capture native Airflow Assets/Datasets as DataHub lineage. See [Native Airflow Assets/Datasets](#native-airflow-assetsdatasets).                                                                                                                          |
| emit_mode                          | ASYNC                | Emit mode for writes to DataHub. `ASYNC` (default) avoids blocking on a synchronous commit per write, reducing GMS load at high volume. Use `SYNC_WAIT`/`SYNC_PRIMARY` for read-after-write or raise-on-failure guarantees.                               |

#### Materializing lineage iolets

By default (`materialize_iolets = true`), the plugin creates the datasets referenced by a task's inlets and outlets if they don't exist in DataHub yet, and records a write operation on each outlet so the dataset shows when the pipeline last updated it. Datasets created this way are placeholders: they have a platform, name, and environment, but no schema or documentation until a source connector ingests the real asset.

Set this to `false` when every table your DAGs touch is already ingested by a source connector (Snowflake, BigQuery, and so on). The plugin then stops creating placeholders, which is useful because SQL parsing can resolve to temporary or mistyped table names you don't want in your catalog. Lineage is emitted either way, but DataHub only renders a lineage edge once both ends exist, so an edge pointing at a dataset that no connector ingests stays hidden — it appears on its own once that dataset is ingested, with no need to re-run the DAG. When the plugin is configured with a DataHub REST connection, it logs a warning for each inlet and outlet that is missing from DataHub.

#### Filtering inlets and outlets

The plugin leaves out two kinds of datasets that never link tasks together:

- `file` datasets that operators report through OpenLineage, which are scratch paths on the worker. Set `capture_ol_file_datasets = true` to keep them. Files you declare as Airflow Assets or manual `inlets`/`outlets` are always kept.
- BigQuery tables in hidden datasets (names starting with `_`), which hold anonymous query results, scripts and sessions. Set `capture_bigquery_temp_datasets = true` to keep them, or change the prefix with `bigquery_temp_table_dataset_prefix`.

To drop other datasets, such as a temporary bucket, set `dataset_pattern`. `dag_filter_str` drops whole DAGs; this drops individual inlets and outlets. Each pattern is a regular expression, case-insensitive, matched from the start of `<platform>:<name>`, for example `gcs:my-bucket/path/file` or `bigquery:my_project.my_dataset.my_table`. End each pattern with `.*` to match everything under a prefix. The key leaves out the environment, so a pattern applies to every `cluster`. If a dataset URN carries a platform instance, which only manually declared inlets/outlets can, the instance is part of the name: `bigquery:my_instance.my_project.my_dataset.my_table`. Denied datasets are removed from the task's inlets and outlets, its runs and its column-level lineage, and are never created in DataHub.

```ini title="airflow.cfg"
[datahub]
dataset_pattern = {"deny": ["gcs:my-tmp-bucket/.*", "s3:my-scratch-bucket/.*"]}
```

On managed Airflow, such as Amazon MWAA or Google Cloud Composer, set the same value as the `datahub.dataset_pattern` configuration override or as an environment variable:

```bash
AIRFLOW__DATAHUB__DATASET_PATTERN='{"deny": ["gcs:my-tmp-bucket/.*", "s3:my-scratch-bucket/.*"]}'
```

## Automatic lineage extraction

To automatically extract lineage information, the plugin builds on top of Airflow's built-in [OpenLineage support](https://airflow.apache.org/docs/apache-airflow-providers-openlineage/stable/supported_classes.html).
As such, we support a superset of the default operators that Airflow/OpenLineage supports.

The SQL-related extractors have been updated to use [DataHub's SQL lineage parser](./sql_parsing.md), which is more robust than the built-in one and uses DataHub's metadata information to generate column-level lineage.

Supported operators:

- `SQLExecuteQueryOperator`, including any subclasses. Most SQL operators inherit from this class.
- `AthenaOperator` and `AWSAthenaOperator`
- `BigQueryOperator` and `BigQueryExecuteQueryOperator`
- `BigQueryInsertJobOperator` (incubating)
- `MySqlOperator`
- `PostgresOperator`
- `RedshiftSQLOperator`
- `SnowflakeOperator` and `SnowflakeOperatorAsync`
- `SqliteOperator`
- `TeradataOperator` (_Note: Teradata uses two-tier `database.table` naming without a schema level_)
- `TrinoOperator`

<!--
These operators are supported by OpenLineage, but we haven't tested them yet:
- `SQLCheckOperator`, `SQLValueCheckOperator`, `SQLThresholdCheckOperator`, `SQLIntervalCheckOperator`, `SQLColumnCheckOperator`, `BigQueryColumnCheckOperator`
- `FTPFileTransmitOperator`
- `GCSToGCSOperator`
- `S3CopyObjectOperator`
- `S3FileTransformOperator`
- `SageMakerProcessingOperator` and `SageMakerProcessingOperatorAsync`
- `SFTPOperator`

There's also a few operators (e.g. BashOperator, PythonOperator) that have custom extractors, but those extractors don't generate lineage.
-->

### Multi-Statement SQL Parsing

When a task executes multiple SQL statements (e.g., `CREATE TEMP TABLE ...; INSERT ... FROM temp_table;`), enable this to parse all statements together and resolve temporary table dependencies. By default (False), only the first statement is parsed.

```ini title="airflow.cfg"
[datahub]
enable_multi_statement_sql_parsing = True  # Default: False
```

**Note:** Use a list of SQL strings (recommended) or semicolon-separated statements in a single string:

## Manual Lineage Annotation

### Using `inlets` and `outlets`

You can manually annotate lineage by setting `inlets` and `outlets` on your Airflow operators. This is useful if you're using an operator that doesn't support automatic lineage extraction, or if you want to override the automatic lineage extraction.

We have a few code samples that demonstrate how to use `inlets` and `outlets`:

- [`lineage_backend_demo.py`](https://github.com/datahub-project/datahub/blob/master/metadata-ingestion-modules/airflow-plugin/src/datahub_airflow_plugin/example_dags/lineage_backend_demo.py)
- [`lineage_backend_taskflow_demo.py`](https://github.com/datahub-project/datahub/blob/master/metadata-ingestion-modules/airflow-plugin/src/datahub_airflow_plugin/example_dags/lineage_backend_taskflow_demo.py) - uses the [TaskFlow API](https://airflow.apache.org/docs/apache-airflow/stable/concepts/taskflow.html)

For more information, take a look at the [Airflow lineage docs](https://airflow.apache.org/docs/apache-airflow/stable/lineage.html).

### Native Airflow Assets/Datasets

You can use native Airflow [Assets](https://airflow.apache.org/docs/apache-airflow/3.0.0/authoring-and-scheduling/assets.html) (previously called Datasets) for data-aware scheduling. The DataHub plugin automatically captures these as lineage when used in `inlets` and `outlets`.

```python
from airflow.sdk.definitions.asset import Asset

s3_input = Asset("s3://my-bucket/input/data.parquet")
bigquery_output = Asset("bigquery://my-project/dataset/result_table")

task = BashOperator(
    task_id="process_data",
    bash_command="echo 'Processing'",
    inlets=[s3_input],
    outlets=[bigquery_output],
)
```

The plugin maps URI schemes to DataHub platforms:

| URI Scheme            | DataHub Platform |
| --------------------- | ---------------- |
| `s3://`, `s3a://`     | s3               |
| `gs://`, `gcs://`     | gcs              |
| `postgresql://`       | postgres         |
| `mysql://`            | mysql            |
| `bigquery://`         | bigquery         |
| `snowflake://`        | snowflake        |
| `file://`             | file             |
| `hdfs://`             | hdfs             |
| `abfs://`, `abfss://` | adls             |

Plain name assets (e.g., from the `@asset` decorator) default to the `airflow` platform.

#### Configuration

```ini title="airflow.cfg"
[datahub]
# Set to false to disable capturing Airflow Assets as lineage (default: true)
capture_airflow_assets = true
```

#### Limitations

Native Airflow Assets have the following limitations compared to using DataHub's `Dataset` or `Urn` entities directly:

1. **No `platform_instance` support**: The URN generated from an Airflow Asset URI cannot include a platform instance. The plugin only extracts the platform, dataset name, and environment from the URI.

2. **Environment uses global plugin config**: All native Airflow Assets use the `cluster` setting from the plugin configuration as their environment. You cannot specify a different environment per asset.

If you need `platform_instance` or per-asset environment control, use the DataHub entity classes instead:

```python
from datahub_airflow_plugin.entities import Dataset

# Full control over URN components
s3_input = Dataset(
    platform="s3",
    name="my-bucket/input/data.parquet",
    env="PROD",
    platform_instance="us-west-2"  # Specify platform instance
)

task = BashOperator(
    task_id="process_data",
    bash_command="echo 'Processing'",
    inlets=[s3_input],
)
```

### Custom Operators

If you have created a [custom Airflow operator](https://airflow.apache.org/docs/apache-airflow/stable/howto/custom-operator.html) that inherits from the BaseOperator class,
when overriding the `execute` function, set inlets and outlets via `context['ti'].task.inlets` and `context['ti'].task.outlets`.
The DataHub Airflow plugin will then pick up those inlets and outlets after the task runs.

You can only set table-level lineage using inlets and outlets. For column-level lineage, you need to write a custom extractor for your custom operator.

```python
class DbtOperator(BaseOperator):
    ...

    def execute(self, context):
        # do something
        inlets, outlets = self._get_lineage()
        # inlets/outlets are lists of either datahub_airflow_plugin.entities.Dataset or datahub_airflow_plugin.entities.Urn
        context['ti'].task.inlets = self.inlets
        context['ti'].task.outlets = self.outlets

    def _get_lineage(self):
        # Do some processing to get inlets/outlets

        return inlets, outlets
```

If you override the `pre_execute` and `post_execute` function, ensure they include the `@prepare_lineage` and `@apply_lineage` decorators respectively. Reference the [Airflow docs](https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/lineage.html#lineage) for more details.

See example implementation of a custom operator using SQL parser to capture table level lineage [here](https://github.com/datahub-project/datahub/blob/master/metadata-ingestion-modules/airflow-plugin/tests/integration/dags/custom_operator_sql_parsing.py)

### Custom SQL Operators with Automatic Lineage

If you're building a custom SQL operator, you have two approaches depending on your needs:

#### Option 1: Inherit from SQLExecuteQueryOperator (Recommended)

**This is the easiest approach** - inherit from Airflow's `SQLExecuteQueryOperator` and you automatically get:

- ✅ OpenLineage support built-in
- ✅ DataHub's enhanced SQL parser (via our SQLParser patch)
- ✅ Column-level lineage extraction
- ✅ No extra code needed

```python
from typing import Any
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

class MyCustomSQLOperator(SQLExecuteQueryOperator):
    """
    Custom SQL operator that inherits OpenLineage support.

    DataHub automatically enhances the SQL parsing with column-level lineage!
    """

    def __init__(self, my_custom_param: str, **kwargs):
        # Add your custom parameters
        self.my_custom_param = my_custom_param
        super().__init__(**kwargs)

    def execute(self, context: Any) -> Any:
        # Add any custom logic before SQL execution
        self.log.info(f"Custom param: {self.my_custom_param}")

        # Parent class handles SQL execution + OpenLineage lineage
        return super().execute(context)
```

**How it works:**

1. `SQLExecuteQueryOperator` already has `get_openlineage_facets_on_complete()` implemented
2. It uses the hook's OpenLineage methods, which internally call `SQLParser`
3. DataHub patches `SQLParser` globally, so all SQL parsing gets enhanced automatically
4. You get column-level lineage without writing any lineage-specific code!

**When to use this:**

- Building operators for standard SQL databases (Postgres, MySQL, Snowflake, BigQuery, etc.)
- Want the simplest integration
- Don't need custom lineage extraction logic

#### Option 2: Implement OpenLineage Interface from Scratch

Only needed if you're building a completely custom operator that doesn't fit the `SQLExecuteQueryOperator` pattern:

```python
from typing import Any, Optional
from airflow.models.baseoperator import BaseOperator

class MyCompletelyCustomOperator(BaseOperator):
    """
    For special cases where SQLExecuteQueryOperator doesn't fit.
    """

    def execute(self, context: Any) -> Any:
        # Your custom SQL execution logic
        pass

    def get_openlineage_facets_on_complete(
        self, task_instance: Any
    ) -> Optional["OperatorLineage"]:
        """
        Implement OpenLineage interface manually.
        DataHub's SQLParser patch still enhances this automatically!
        """
        from airflow.providers.openlineage.sqlparser import SQLParser

        hook = self.get_db_hook()
        parser = SQLParser(
            dialect=hook.get_openlineage_database_dialect(hook.get_connection(self.conn_id)),
            default_schema=hook.get_openlineage_default_schema(),
        )

        # This uses DataHub's patched SQLParser - column lineage included!
        return parser.generate_openlineage_metadata_from_sql(
            sql=self.sql,
            hook=hook,
            database_info=hook.get_openlineage_database_info(hook.get_connection(self.conn_id)),
        )
```

**When to use this:**

- Building a very specialized operator from scratch
- Need complete control over lineage extraction
- The standard SQLExecuteQueryOperator pattern doesn't apply

**Key Point:** DataHub patches `SQLParser.generate_openlineage_metadata_from_sql()` globally at import time, so **any operator** using OpenLineage's SQLParser automatically gets DataHub's enhanced parsing with column-level lineage!

### Alternative: Custom Operators with Manual Lineage

If you prefer not to use OpenLineage, or are on older Airflow versions, you can manually extract and set lineage using DataHub's SQL parser:

```python
from typing import Any, List, Tuple, Union
from airflow.models.baseoperator import BaseOperator
from datahub_airflow_plugin._config import get_enable_multi_statement
from datahub_airflow_plugin._sql_parsing_common import parse_sql_with_datahub
from datahub_airflow_plugin.entities import Urn

class CustomSQLOperator(BaseOperator):
    def __init__(self, sql: Union[str, List[str]], database: str, **kwargs: Any):
        super().__init__(**kwargs)
        self.sql = sql
        self.database = database

    def execute(self, context: Any) -> Any:
        # Execute SQL
        # ...

        # Extract and set lineage
        inlets, outlets = self._get_lineage()
        context["ti"].task.inlets = inlets
        context["ti"].task.outlets = outlets

    def _get_lineage(self) -> Tuple[List, List]:
        # Get multi-statement config flag
        enable_multi_statement = get_enable_multi_statement()

        # Parse SQL with multi-statement support
        # Handles both string and list of SQL statements
        sql_parsing_result = parse_sql_with_datahub(
            sql=self.sql,
            platform="postgres",  # your platform
            default_database=self.database,
            env="PROD",
            default_schema=None,
            graph=None,
            enable_multi_statement=enable_multi_statement,
        )

        inlets = [Urn(table) for table in sql_parsing_result.in_tables]
        outlets = [Urn(table) for table in sql_parsing_result.out_tables]
        return inlets, outlets
```

See [full example](https://github.com/datahub-project/datahub/blob/master/metadata-ingestion-modules/airflow-plugin/tests/integration/dags/custom_operator_sql_parsing.py).

## Cleanup obsolete pipelines and tasks from DataHub

There might be a case where the DAGs are removed from the Airflow but the corresponding pipelines and tasks are still there in the DataHub, let's call such pipelines and tasks, `obsolete pipelines and tasks`

Following are the steps to cleanup them from the datahub:

- create a DAG named `Datahub_Cleanup`, i.e.

```python
from datetime import datetime

from airflow import DAG
from airflow.operators.bash import BashOperator

from datahub_airflow_plugin.entities import Dataset, Urn

with DAG(
    "Datahub_Cleanup",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
    catchup=False,
) as dag:
    task = BashOperator(
        task_id="cleanup_obsolete_data",
        dag=dag,
        bash_command="echo 'cleaning up the obsolete data from datahub'",
    )

```

- ingest this DAG, and it will remove all the obsolete pipelines and tasks from the DataHub based on the `cluster` value set in the `airflow.cfg`

## Get all dataJobs associated with a dataFlow

If you are looking to find all tasks (aka DataJobs) that belong to a specific pipeline (aka DataFlow), you can use the following GraphQL query:

```graphql
query {
  dataFlow(urn: "urn:li:dataFlow:(airflow,db_etl,prod)") {
    childJobs: relationships(
      input: { types: ["IsPartOf"], direction: INCOMING, start: 0, count: 100 }
    ) {
      total
      relationships {
        entity {
          ... on DataJob {
            urn
          }
        }
      }
    }
  }
}
```

## Emit Lineage Directly

If you can't use the plugin or annotate inlets/outlets, you can also emit lineage using the `DatahubEmitterOperator`.

Reference [`lineage_emission_dag.py`](https://github.com/datahub-project/datahub/blob/master/metadata-ingestion-modules/airflow-plugin/src/datahub_airflow_plugin/example_dags/lineage_emission_dag.py) for a full example.

In order to use this example, you must first configure the DataHub hook. Like in ingestion, we support a DataHub REST hook and a Kafka-based hook. See the plugin configuration for examples.

## Debugging

### Missing lineage

If you're not seeing lineage in DataHub, check the following:

- Validate that the plugin is loaded in Airflow. Go to Admin -> Plugins and check that the DataHub plugin is listed.
- It should also print a log line like `INFO  [datahub_airflow_plugin.datahub_listener] DataHub plugin using DataHubRestEmitter: configured to talk to <datahub_url>` during Airflow startup, and the `airflow plugins` command should list `datahub_plugin` with a listener enabled.
- If using the plugin's automatic lineage, ensure that the `enable_extractors` config is set to true and that automatic lineage is supported for your operator.
- If using manual lineage annotation, ensure that you're using the `datahub_airflow_plugin.entities.Dataset` or `datahub_airflow_plugin.entities.Urn` classes for your inlets and outlets.

### Incorrect URLs

If your URLs aren't being generated correctly (usually they'll start with `http://localhost:8080` instead of the correct hostname), you may need to set the webserver `base_url` config.

```ini title="airflow.cfg"
[webserver]
base_url = http://airflow.mycorp.example.com
```

### TypeError ... missing 3 required positional arguments

If you see errors like the following:

```shell
ERROR - on_task_instance_success() missing 3 required positional arguments: 'previous_state', 'task_instance', and 'session'
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/datahub_airflow_plugin/datahub_listener.py", line 124, in wrapper
    f(*args, **kwargs)
TypeError: on_task_instance_success() missing 3 required positional arguments: 'previous_state', 'task_instance', and 'session'
```

The solution is to upgrade `acryl-datahub-airflow-plugin>=0.12.0.4` or upgrade `pluggy>=1.2.0`. See this [PR](https://github.com/datahub-project/datahub/pull/9365) for details.

### Orphan dataset URNs containing `.None.`

If you see a warning in the Airflow task logs that looks like this:

```text
WARNING  datahub_airflow_plugin._datahub_ol_adapter: OpenLineage Dataset name
'mydb.None.public.events' contained 'None'/empty segments; sanitized to
'mydb.public.events' before producing DataHub URN. Likely upstream bug in the
producer (unset field interpolated into an f-string).
```

it means an upstream OpenLineage producer emitted a Dataset whose `name` had
the literal string `"None"` baked into one of its dotted segments. This
happens when the producer builds the name with an f-string and one of the
interpolated fields is Python `None` (which renders as the four-character
string `"None"`). One example you can see in upstream source is Apache
Airflow's `S3ToRedshiftOperator.get_openlineage_facets_on_complete`, which
constructs the name as
[`f"{database}.{self.schema}.{self.table}"`](https://github.com/apache/airflow/blob/67b71d376454cc95cf2f5bb17e0f4edb0e05f480/providers/amazon/src/airflow/providers/amazon/aws/transfers/s3_to_redshift.py#L260)
with no `None` guard — but any producer (Airflow operator, dbt adapter, Spark
listener, custom integration) following the same pattern can hit it.

The DataHub plugin sanitizes these names automatically before emitting the URN
so that lineage stitches correctly to whatever your DataHub native Redshift /
Snowflake / etc. ingestion source produced for the same physical table — no
action is required for new lineage. The warning is informational and serves as
a hint that you may want to file an upstream issue against the offending
provider.

The warning fires **at most once per worker process**. The first buggy name
the sanitiser sees produces a log line, and subsequent occurrences (same name
or different) stay silent for the lifetime of that process — the warning is
an alarm, not a stream, and additional lines would only enumerate the
symptom. To find the full set of orphans, search DataHub for `.None.` or
empty segments using the cleanup commands below.

If a Dataset name consists **entirely** of `"None"` and empty segments
(pathological producer output), the plugin keeps the original name to avoid
emitting an empty-name URN and the warning text changes accordingly:

```text
WARNING  datahub_airflow_plugin._datahub_ol_adapter: OpenLineage Dataset name
'None.None' had only 'None'/empty segments; kept original to avoid emitting
an empty URN. The resulting DataHub URN will literally contain 'None.None'
in its name field — search DataHub for that substring to find orphans and
report the upstream producer.
```

Such URNs remain reachable in the catalog (so they can be soft-deleted with
the cleanup commands below) rather than disappearing into an unsearchable
empty-name URN.

If you have **historical** orphan URNs in DataHub from before upgrading the
plugin (datasets whose `name` still contains a literal `.None.` segment), they
will not be cleaned up automatically. **Always preview the deletions first**
with `--dry-run`, then drop the flag to actually delete:

```shell
# 1. Preview the URNs that would be soft-deleted (no destructive action).
datahub delete --platform redshift --soft --query "*.None.*" --dry-run

# 2. Once you've confirmed the list is correct, run the same command
#    without --dry-run to perform the soft-delete.
datahub delete --platform redshift --soft --query "*.None.*"
```

Soft-deleted entities can be restored with
`datahub delete undo-by-filter --platform redshift` if you delete more than
you intended.

### Scheduler stalling

For extremely large Airflow deployments with thousands of tasks, you may see issues where the plugin interferes with the performance of the Airflow scheduler. In those cases, you can set the `DATAHUB_AIRFLOW_PLUGIN_RUN_IN_THREAD_TIMEOUT=0` environment variable. This makes the DataHub plugin run fully in background threads, but can cause us to miss some metadata if the scheduler shuts down soon after processing a task.

### Disabling the DataHub Plugin

There are two ways to disable the DataHub Plugin:

#### 1. Disable via Configuration

Set the `datahub.enabled` configuration property to `False` in the `airflow.cfg` file and restart the Airflow environment to reload the configuration and disable the plugin.

```ini title="airflow.cfg"
[datahub]
enabled = False
```

#### 2. Disable via Environment Variable (Kill-Switch)

If a restart is not possible and you need a faster way to disable the plugin, you can use the kill-switch. Set the `AIRFLOW_VAR_DATAHUB_AIRFLOW_PLUGIN_DISABLE_LISTENER` environment variable to `true`. This ensures that the listener won't process anything.

```shell
export AIRFLOW_VAR_DATAHUB_AIRFLOW_PLUGIN_DISABLE_LISTENER=true
```

This will immediately disable the plugin without requiring a restart.

:::note Why Environment Variable Instead of Airflow Variable?
The plugin uses environment variables instead of Airflow's `Variable.get()` because listener hooks are called during SQLAlchemy's `after_flush` event (before the main transaction commits). Calling `Variable.get()` in this context creates a nested database session that can interfere with the outer transaction and cause data loss, such as missing TaskInstanceHistory records for retried tasks.
:::

## Compatibility

We try to support Airflow releases for ~2 years after their release. This is a best-effort guarantee - it's not always possible due to dependency / security issues cropping up in older versions.

We no longer officially support Airflow 2.x — Apache has EOLed the 2.x line. To keep using the plugin against Airflow 2.x, pin `acryl-datahub-airflow-plugin <= 1.6.0` (the last release with Airflow 2 support). Airflow 3.0+ is supported.

We previously had two implementations of the plugin - v1 and v2. The v2 plugin is now the default, and the v1 plugin has since been removed. The v1 plugin had many limitations, chiefly that it does not support automatic lineage extraction. Docs for the v1 plugin can be accessed in our [docs archive](https://archive.docs.datahub.com/docs/0.15.0/lineage/airflow#datahub-plugin-v1).

All recent versions require Python 3.10+.

- Airflow 1.10.x, use acryl-datahub-airflow-plugin <= 0.9.1.0 (v1 plugin).
- Airflow 2.0.x, use acryl-datahub-airflow-plugin <= 0.11.0.1 (v1 plugin).
- Airflow 2.2.x, use acryl-datahub-airflow-plugin <= 0.14.1.5 (v2 plugin).
- Airflow 2.3 - 2.4.3, use acryl-datahub-airflow-plugin <= 1.0.0 (v2 plugin).
- Airflow 2.5 and 2.6, use acryl-datahub-airflow-plugin <= 1.1.0.4 (v2 plugin).
- Airflow 2.7 - 2.10, use acryl-datahub-airflow-plugin <= 1.6.0 (v2 plugin; the last release with Airflow 2 support).
- Airflow 3.0+ is supported by the current release.

DataHub also previously supported an Airflow [lineage backend](https://airflow.apache.org/docs/apache-airflow/2.2.0/lineage.html#lineage-backend) implementation. The lineage backend functionality was pretty limited - it did not support automatic lineage extraction, did not capture task failures, and did not work in AWS MWAA - and so it has been removed from the codebase. The [documentation for the lineage backend](https://archive.docs.datahub.com/docs/0.10.5/lineage/airflow#using-datahubs-airflow-lineage-backend-deprecated) has been archived.

## Changelog

### Next

- _Fixes_:
  - Installing with Apache Airflow's constraint file needs those two pins relaxed first.
    `click` is `>=8.3.3` and `anyio` is `>=4.14.2`, and Airflow's file still pins older
    releases, so a constrained install cannot resolve until the file is rewritten. See
    [Installing with Airflow constraints](#installing-with-airflow-constraints).

`acryl-datahub-airflow-plugin` is published to PyPI on **every** DataHub release and carries that
release's version, so `acryl-datahub-airflow-plugin==1.7.0.10` is the plugin built from DataHub
`v1.7.0.10`. Most releases change nothing in the plugin, so only the versions that changed it are
listed here. A version that is not listed behaves the same as the nearest older listed version,
and picking the newest published version is always safe.

Entries start at `1.3.1.5`. For anything earlier, and for the full migration detail behind the
breaking changes below, see [Breaking Changes](../how/breaking-changes.md).

### Next

_Changes_

- OpenLineage references to S3, GCS, Azure (`abfs`, `wasb`) and DBFS paths now use the same platform and name as DataHub's storage sources and the Spark agent: `gs://my-bucket` plus `path/file` becomes `urn:li:dataset:(urn:li:dataPlatform:gcs,my-bucket/path/file,PROD)` instead of a `gs` dataset without the bucket. Set `normalize_object_storage_urns = false` to keep the old URNs for OpenLineage references; the Asset changes below do not depend on it. Airflow Assets on `abfs://` and `abfss://` now use the `abs` platform instead of `adls`. See [Breaking Changes](../how/breaking-changes.md).
- New `dataset_pattern` option to drop individual inlets and outlets by `<platform>:<name>` pattern. See [Filtering inlets and outlets](#filtering-inlets-and-outlets).
- `file` datasets reported through OpenLineage (worker scratch paths) are no longer captured by default; set `capture_ol_file_datasets = true` to keep them. BigQuery tables in hidden `_`-prefixed datasets are dropped by default; set `capture_bigquery_temp_datasets = true` to keep them.
- Airflow Assets with a `file:///` URI keep the leading slash in the dataset name (`file:///tmp/x` becomes `/tmp/x`, not `tmp/x`), matching OpenLineage and the Spark agent.
- DataFlows now carry a `schedule` custom property with the DAG's schedule as the Airflow UI shows it (for example `0 6 * * *`, `2:00:00` or `Asset`; `None` for an unscheduled DAG). The value is the same whether the DAG-run or the task hooks emitted the DataFlow. If the schedule cannot be resolved, for example a custom timetable that isn't registered through an Airflow plugin, the property is omitted and a warning is logged ([#20101](https://github.com/datahub-project/datahub/pull/20101)).

### Versions 1.7.0 through 1.7.0.10

No plugin changes. Every `1.7.x` release published so far ships the same plugin code as
`1.6.0.16`.

### Version 1.6.0.16

- _Fixes_:
  - Kafka emitter configuration is parsed with pydantic v2's `model_validate` instead of the
    removed v1 `parse_obj`, so `DatahubKafkaHook` and the listener's Kafka sink work on pydantic 2
    ([#18432](https://github.com/datahub-project/datahub/pull/18432))

### Version 1.6.0.9

- _Changes_:
  - **The default emit mode changed from synchronous to `ASYNC`**
    ([#18014](https://github.com/datahub-project/datahub/pull/18014)) to reduce GMS load at high
    volume. Under async, `emit()` no longer raises on a rejected write and there is no
    read-after-write guarantee. To restore the previous behaviour, set
    `emit_mode = SYNC_PRIMARY` under `[datahub]` in `airflow.cfg`.

### Version 1.6.0.6

- _Changes_:
  - `AssetAlias` is recognised as lineage on Airflow 3+
    ([#17055](https://github.com/datahub-project/datahub/pull/17055)). An alias names an asset that
    only resolves at task runtime — for example when the URI contains a Jinja template — so
    asset-triggered DAGs now produce lineage that previously went missing.

### Version 1.6.0.1

- _Major changes_:
  - **Airflow 2.x support dropped; Airflow 3.0+ required**
    ([#17439](https://github.com/datahub-project/datahub/pull/17439)). The `apache-airflow` floor
    is `>=3.0.0`. The standalone `openlineage-airflow` package is no longer used at all — the
    plugin always uses `apache-airflow-providers-openlineage>=2.1.0`, now an unconditional
    dependency — so drop `openlineage-airflow` from your constraints if it is pinned. The
    `[airflow2]` extra is gone and `[airflow3]` is retained as a no-op, so existing
    `pip install 'acryl-datahub-airflow-plugin[airflow3]'` commands keep working. The
    `taskinstance` value for `datajob_url_link` is no longer accepted; use `tasks` or `grid`. The
    `patch_snowflake_schema` option was removed and is silently ignored. To stay on Airflow 2.7
    through 2.10, pin `acryl-datahub-airflow-plugin <= 1.6.0`.

### Version 1.5.0.19

- _Fixes_:
  - OpenLineage dataset names containing a literal `None` segment are sanitized before the URN is
    built ([#17191](https://github.com/datahub-project/datahub/pull/17191)). Some Airflow providers
    assemble names with f-strings and no `None` guard — `S3ToRedshiftOperator` with an unset
    `schema`, for example — which produced orphan `urn:li:dataset:(...,db.None.schema.tbl,PROD)`
    entities instead of stitching to the URN native ingestion emits. Orphans created before
    upgrading are not removed automatically; see
    [Orphan dataset URNs containing `.None.`](#orphan-dataset-urns-containing-none).

### Version 1.5.0.10

- _Fixes_:
  - The `MappedOperator` import falls back to
    `airflow.serialization.definitions.mappedoperator` when `airflow.models.mappedoperator` is
    absent ([#16963](https://github.com/datahub-project/datahub/pull/16963)), so the plugin loads
    on Airflow 3.2+, which moved the class.

### Version 1.5.0.8

- _Fixes_:
  - Hardened the import fallback in the Teradata OpenLineage wrapper
    ([#16868](https://github.com/datahub-project/datahub/pull/16868))

### Version 1.4.0.4

- _Changes_:
  - Multi-statement SQL parsing for lineage extraction
    ([#16235](https://github.com/datahub-project/datahub/pull/16235)). See
    [Multi-Statement SQL Parsing](#multi-statement-sql-parsing).
  - The `datahub-file` connection type is registered with the Airflow provider
    ([#16168](https://github.com/datahub-project/datahub/pull/16168)), so `SynchronizedFileHook` is
    selectable in the Airflow connection UI and defaults to the `datahub_file_default` connection.

### Version 1.4.0

- _Fixes_:
  - Dataset URNs in lineage use the configured DataHub cluster environment rather than a default
    ([#16040](https://github.com/datahub-project/datahub/pull/16040)), so lineage stitches to the
    same URNs as native ingestion when `cluster` is not `PROD`.

### Version 1.3.1.8

- _Major changes_:
  - **Python 3.9 support dropped; Python 3.10+ required**
    ([#15984](https://github.com/datahub-project/datahub/pull/15984))
- _Changes_:
  - Airflow Asset support on both Airflow 2 and 3
    ([#15947](https://github.com/datahub-project/datahub/pull/15947)). See
    [Native Airflow Assets/Datasets](#native-airflow-assetsdatasets).
- _Fixes_:
  - The Airflow 2.x kill switch reads the environment variable
    `AIRFLOW_VAR_DATAHUB_AIRFLOW_PLUGIN_DISABLE_LISTENER` instead of an Airflow Variable
    ([#15930](https://github.com/datahub-project/datahub/pull/15930)), because Airflow 3 forbids
    database access during listener initialization.

### Version 1.3.1.7

- _Fixes_:
  - The DataFlow is emitted from the task handler
    ([#15875](https://github.com/datahub-project/datahub/pull/15875)), so distributed Airflow 3
    deployments — where the scheduler and workers run in separate processes — emit the pipeline
    rather than only its tasks.

### Version 1.3.1.5

- _Changes_:
  - **Airflow 3.x support** ([#13790](https://github.com/datahub-project/datahub/pull/13790)). The
    plugin detects the Airflow version and selects the matching listener, moves to
    `API__BASE_URL` where Airflow 3 renamed `WEBSERVER__BASE_URL`, and replaces the
    operator-specific extractors with a unified SQLParser patch for lineage. SubDAGs are gone in
    Airflow 3; use TaskGroups.
  - Teradata operator support
    ([#15418](https://github.com/datahub-project/datahub/pull/15418))

## Additional references

Related DataHub videos:

- [Airflow Lineage](https://www.youtube.com/watch?v=3wiaqhb8UR0)
- [Airflow Run History in DataHub](https://www.youtube.com/watch?v=YpUOqDU5ZYg)