Airflow Lineage Integration

Lineage from Airflow DAGs is not appearing in DataHub, or the integration between Airflow and DataHub is not working correctly. This typically occurs due to missing Airflow plugin configuration, incorrect DataHub connection settings, or unsupported operator types.

You Might Be Asking:

• How do I integrate Airflow with DataHub?
• Why isn't my DAG lineage showing up?
• Which Airflow operators are supported?

1. Install DataHub Airflow plugin:

# Install the plugin
pip install 'acryl-datahub[airflow]'

# Or add to requirements.txt
acryl-datahub[airflow]>=0.10.0

2. Configure Airflow connection to DataHub:

# In airflow_local_settings.py or airflow.cfg
DATAHUB_CONN_ID = "datahub_rest_default"

# Configure the connection in Airflow UI:
# Conn Id: datahub_rest_default
# Conn Type: HTTP
# Host: http://datahub-gms:8080
# Extra: {"timeout": 60}

# Or via environment variable
AIRFLOW_CONN_DATAHUB_REST_DEFAULT='http://datahub-gms:8080'

3. Enable DataHub lineage in DAG:

Example for Snowflake. This can be adapted for other data sources/connectors.

from airflow import DAG
from airflow.operators.python import PythonOperator
from datahub.integrations.airflow.operators import DatahubEmitterOperator
from datahub.emitter.mce_builder import make_dataset_urn

# Define lineage using inlets and outlets
with DAG(
    dag_id="example_dag",
    schedule_interval="@daily",
    catchup=False
) as dag:

    transform_task = PythonOperator(
        task_id="transform_data",
        python_callable=my_transform_function,
        # Define lineage
        inlets=[
            make_dataset_urn("snowflake", "db.schema.source_table")
        ],
        outlets=[
            make_dataset_urn("snowflake", "db.schema.target_table")
        ]
    )

4. Use DataHub-aware operators:

Example for Snowflake. This can be adapted for other data sources/connectors.

from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator
from datahub.integrations.airflow.operators import DatahubSnowflakeOperator

# Automatically extracts lineage
task = DatahubSnowflakeOperator(
    task_id="snowflake_transform",
    sql="""
        INSERT INTO target_table
        SELECT * FROM source_table
        WHERE date = '{{ ds }}'
    """,
    snowflake_conn_id="snowflake_default",
    datahub_conn_id="datahub_rest_default",
    dag=dag
)

5. Configure plugin settings:

# In airflow_local_settings.py
DATAHUB_CONFIG = {
    "enabled": True,
    "datahub_conn_id": "datahub_rest_default",
    "cluster": "prod",
    "capture_ownership_info": True,
    "capture_tags_info": True,
    "capture_executions": True,
    "capture_lineage": True,
}

The DataHub Airflow plugin automatically emits metadata for supported operators. For custom operators, manually define inlets and outlets. DAG-level lineage creates relationships between all tasks and their data dependencies.

• Airflow Integration
• Lineage Backend