Integrating with Apache Airflow (OpenLineage Provider)

Prerequisites

To use the OpenLineage Provider integration, your environment must meet the following requirements:

• Apache Airflow 2.9 or later. This version is required because essential features in the OpenLineage Provider are only available from Airflow 2.9.
• OpenLineage Python client 1.44 or later (required for API key authentication).

If you use an older version of Airflow, you cannot use the OpenLineage Provider. In that case, you must create the integration by using the OpenLineage API directly. For more information, see OpenLineage integration for off-platform assets.

Setting up integration credentials and endpoints

Before you configure Airflow, you must set up the authentication credentials and API endpoints that are required to connect to Data Observability:

1. Create an API key in the IBM Cloud console.

2. Log in to watsonx.data integration and copy the project ID where you want to observe job runs. You can find the project ID in the Manage tab of the project. If you don't have a project, create one and copy its ID. You must have an existing project to receive the job metadata.

3. Configure OpenLineage Provider to send data to the API endpoint. The API endpoint uses the following format for both single and batch event submission:

   https://{watsonx_hostname}/data-obs/v1/lineage

Configuring Apache Airflow

1. Install the OpenLineage Provider package (minimum version 2.1.0):

   pip install apache-airflow-providers-openlineage>=2.1.0
   

2. Install the OpenLineage Python client (minimum version 1.44.0):

   pip install openlineage-python>=1.44.0
   

3. In Airflow, verify that openlineage_disabled is set to False. You can check this setting in Admin > Configurations.

4. Optional: To send events only from selected DAGs, set selective_enable to True. If selective_enable is set to False (the default), all DAGs send lineage information.

5. Configure the OpenLineage transport settings (for example, in your Airflow configuration file). The following example specifies the payload destination and authentication method. Replace:

   • your_api_key with your IBM Cloud API key.
   • https://watsonx_hostname with the URL of the server.
     
   • IAM Identity Services API endpoint (for example, https://iam.cloud.ibm.com/identity/token). For more details, see Apache Airflow OpenLineage Provider.

   {
   "type": "http",
     "url": "https://watsonx_hostname",
     "endpoint": "data-obs/v1/lineage",
     "auth": {
       "type": "jwt",
       "apiKey": "your_api_key",
       "tokenEndpoint": "https://iam.cloud.ibm.com/identity/token",
       "grantType": "urn:ibm:params:oauth:grant-type:apikey",
       "responseType": "cloud_iam"
     }
   }
   

6. Add one of the following tags to your DAG definition to assign it to a specific project or space:

   • ibm_project_id:<your_project_id>
   • ibm_space_id:<your_space_id>
     By using these tags, you specify where your lineage events are stored and displayed in watsonx.data integration. For example, if you have job A that needs to appear in Project A and job B that needs to appear in Project B, you specify different project_id values in the DAG tags for each job's events. By using tags, you can set the project on a per DAG basis.

7. Optional: To enable sending lineage data for chosen DAGs only:

   • If you set selective_enable to True, wrap your DAG with enable_lineage ().
     If selective_enable is set to False, all DAGs send lineage information by default.

The following example shows a DAG created so that it is displayed in the specified project. It shows how you can set the target project on a per-DAG basis.

from airflow.providers.openlineage.utils.selective_enable import enable_lineage
from datetime import datetime
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator

def simple_python_task():
    print("Hello! This is your hourly Python task running successfully.")

with enable_lineage(DAG(
    dag_id='example_openlineage_dag',
    schedule='@hourly',
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=['ibm_project_id:<project_id>', 'other_tag'],
)) as dag:

    task_bash = BashOperator(
        task_id='print_date_bash',
        bash_command='date',
    )

    task_python = PythonOperator(
        task_id='print_message_python',
        python_callable=simple_python_task,
    )

    task_bash >> task_python