Airbyte Apache Airflow Provider

5.4.0 · active · verified Sat Apr 11

The `apache-airflow-providers-airbyte` package provides Apache Airflow operators and sensors to interact with Airbyte, an open-source data integration platform. It enables users to trigger and monitor Airbyte synchronization jobs directly from Airflow DAGs. The current version is 5.4.0, supporting Airflow >=2.11.0 and Python >=3.10, and it maintains a regular release cadence with ongoing development.

Warnings

breaking Authentication mechanism for Airbyte connections changed in provider version 4.0.0. It now uses `client_id` and `client_secret` instead of `login` and `password`. The `host` parameter for the Airflow Airbyte connection must be a Fully Qualified Domain Name (FQDN) including schema (e.g., `https://my.company:8000/airbyte/v1/`). The `api_type` parameter was also removed.
Fix: Update your Airflow Airbyte connection configuration to use `client_id` and `client_secret` for authentication and ensure the host is a complete FQDN. Remove `api_type` parameter if present.
breaking The minimum required Apache Airflow version has increased over time. For provider version 5.4.0, Airflow 2.11.0+ is required. Additionally, provider version 5.4.0 removed the `polling_interval` parameter from `AirbyteJobSensor`, favoring `poke_interval`.
Fix: Ensure your Airflow environment is at least 2.11.0. Replace `polling_interval` with `poke_interval` in `AirbyteJobSensor`.
breaking Provider version 2.0.0 (and subsequently 2.1.0+) required Airflow 2.1.0+ due to the removal of the `apply_default` decorator. If upgrading the provider on an older Airflow, this could lead to automatic Airflow package upgrades and require a `airflow upgrade db` command.
Fix: Upgrade Airflow to at least 2.1.0 (preferably the latest compatible version for your provider) before upgrading the Airbyte provider. Run `airflow upgrade db` if automatic upgrade occurs.
gotcha The `AirbyteTriggerSyncOperator` is not idempotent by design. Re-triggering the operator may initiate a new sync job in Airbyte, depending on Airbyte's configuration for the connection. Users should be aware of the Airbyte source/destination sync mode.
Fix: Understand Airbyte's sync configuration (e.g., incremental vs. full refresh) for the connection being triggered and design your DAGs accordingly to handle potential non-idempotent behavior.
gotcha The Airbyte operator in this provider is primarily designed to work with Airbyte self-managed instances (using its internal Config API). For orchestrating Airbyte Cloud, it's generally recommended to use Airflow's generic HTTP operators to interact with the newer Airbyte API directly.
Fix: If using Airbyte Cloud, consider implementing `airflow.providers.http.operators.http.SimpleHttpOperator` for more direct API calls to the Airbyte Cloud API instead of the dedicated Airbyte provider operators.
gotcha To prevent conflicts and ensure Airflow maintains control, it's highly recommended to set the replication frequency for Airbyte connections triggered by Airflow to 'Manual' within the Airbyte UI.
Fix: In the Airbyte UI, for any connection orchestrated by Airflow, set its replication frequency to 'Manual'.

Install

pip install apache-airflow-providers-airbyte Base Installation
pip install apache-airflow-providers-airbyte[http] With HTTP dependencies (often needed)

Imports

AirbyteTriggerSyncOperator

from airflow.providers.airbyte.operators.airbyte import AirbyteTriggerSyncOperator

AirbyteJobSensor

from airflow.providers.airbyte.sensors.airbyte import AirbyteJobSensor

AirbyteHook
```
from airflow.providers.airbyte.hooks.airbyte import AirbyteHook
```
Typically used internally by operators/sensors or for custom interactions, less common directly in DAGs.

Quickstart

This quickstart DAG demonstrates how to trigger and monitor an Airbyte synchronization job using the `AirbyteTriggerSyncOperator` and `AirbyteJobSensor`. Before running, you need to: 1. Install the provider: `pip install apache-airflow-providers-airbyte[http]`. 2. Configure an Airflow 'Airbyte' connection (e.g., `AIRBYTE_CONNECTION_ID`) pointing to your Airbyte instance's API (e.g., `http://localhost:8001`). 3. Obtain the UUID of the specific Airbyte connection you wish to sync from the Airbyte UI (this is `AIRBYTE_SYNC_CONNECTION_ID`). 4. Set `AIRBYTE_CONN_ID` and `AIRBYTE_SYNC_CONN_ID` as environment variables or replace them directly in the DAG code.

import os
from datetime import datetime, timedelta

from airflow import DAG
from airflow.providers.airbyte.operators.airbyte import AirbyteTriggerSyncOperator
from airflow.providers.airbyte.sensors.airbyte import AirbyteJobSensor
from airflow.utils.dates import days_ago

AIRBYTE_CONNECTION_ID = os.environ.get('AIRBYTE_CONN_ID', 'your_airflow_airbyte_connection_id')
AIRBYTE_SYNC_CONNECTION_ID = os.environ.get('AIRBYTE_SYNC_CONN_ID', 'your_airbyte_workspace_connection_id')

with DAG(
    dag_id='example_airbyte_sync_dag',
    start_date=days_ago(1),
    schedule_interval=None,
    catchup=False,
    tags=['airbyte', 'example'],
    dagrun_timeout=timedelta(minutes=60),
    default_args={
        'owner': 'airflow',
    }
) as dag:
    trigger_airbyte_sync = AirbyteTriggerSyncOperator(
        task_id='trigger_airbyte_connection_sync',
        airbyte_conn_id=AIRBYTE_CONNECTION_ID,
        connection_id=AIRBYTE_SYNC_CONNECTION_ID, # This is the UUID of the Airbyte connection to trigger
        asynchronous=True, # Recommended to use with a sensor for long-running jobs
    )

    monitor_airbyte_sync = AirbyteJobSensor(
        task_id='monitor_airbyte_connection_sync',
        airbyte_conn_id=AIRBYTE_CONNECTION_ID,
        airbyte_job_id=trigger_airbyte_sync.output,
        poke_interval=5, # Check every 5 seconds
        timeout=3600, # Timeout after 1 hour
    )

    trigger_airbyte_sync >> monitor_airbyte_sync

view raw JSON →