Orchestrate Airbyte using Apache Airflow

In today’s data-driven world, there are numerous data sources, each with its structure, frequency, and format. Extracting and transferring data from these sources to a data warehouse can be a challenging and time-consuming process. That’s where Airbyte comes in as an open-source data integration platform that makes it easy to connect to various data sources and replicate data in a standardized format to a target destination.

However, scheduling Airbyte jobs manually can be tedious, error-prone, and time-consuming. This is where Apache Airflow comes in. Airflow is a platform to programmatically author, schedule, and monitor workflows. It lets you define your workflows as code, execute them automatically, and monitor their progress.

In this blog summary, we will discuss how to orchestrate Airbyte with Airflow to create a reliable and automated data integration pipeline. We will cover the following steps:

1. Setting up Airbyte: We will cover how to set up Airbyte and how to create a connection to the desired data source(s) and destination(s).
2. Setting up Airflow: We will cover how to set up Airflow and how to create an Airflow DAG (Directed Acyclic Graph) to schedule the Airbyte job.
3. Creating the DAG: We will walk through the process of creating an Airflow DAG that uses an Airbyte operator to execute an Airbyte job.
4. Configuring the DAG: We will discuss how to configure the DAG with parameters such as scheduling frequency, time zone, and email notifications.
5. Monitoring the DAG: We will cover how to monitor the DAG using the Airflow web interface and how to troubleshoot any issues.

Following these steps, you can orchestrate Airbyte with Airflow to create a reliable and automated data integration pipeline that saves time, reduces errors, and simplifies the data integration process.

Step 1:- Create a Connection using Airbyte UI

• I am assuming that you have installed Airbyte on your system and started the webserver and scheduler.

Open Browser → Type Localhost:8000 → New connection

Airbyte Homepage

Step 2:- Add a Source

• I have added ‘Chargebee’ as the source

Chargebee Configuration

Step 3:- Add a Destination

• I took “AWS S3” as my destination

AWS S3 Configuration

Step 4:- Start the Sync

→ Click on sync now

Connection created successfully

Step 5:- Set up the Frequency

• Set up the frequency of this connection to sync as per your requirement.

Replication Frequency

Great job on successfully setting up the Airbyte connection!👌Now, by orchestrating it with Airflow, you can automate the process of data extraction, transformation, and loading. With Airflow, you can easily schedule and monitor the data pipeline to ensure smooth and reliable data transfers. So let’s get started and streamline your data pipeline with Airflow.

Before we proceed, please make sure that Airflow is installed on your system. If you haven’t installed it yet, you can follow the below documentation to install Airflow locally on your System to get it set up.

Install Airflow on Windows without Docker or Virtual Box in 5 mins

Once you have Airflow installed, we can move forward with setting up the orchestration of the Airbyte connection.

Step 1:- Create an Airbyte connection in Apache Airflow

A) Open Airflow UI → Click Admin → Click Connections

Airflow UI

B) Add a connection

• Choose the Connection type:- HTTP

The Airbyte API uses HTTP, so we’ll need to create an HTTP Connection.

• Host:- Enter the URL of your Airbyte Instance
• Port:- 8001 (8000)

Choose Connection type — HTTP

Step 2:- Retrieving the Airbyte Connection ID

We’ll need the Airbyte Connection ID so our Airflow DAG knows which Airbyte Connection to trigger.

Copy the Connection ID as highlighted

Step 3:- Create a simple Airflow DAG to run an Airbyte Sync Job

I am using Sensor to monitor my Airbyte sync as Sensors do not occupy an Airflow worker slot, so this helps reduce Airflow load.

a) Install Airbyte Provider on your instance

→ Run

pip install apache-airflow-providers-airbyte[http]

b) DAG script

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

with DAG(dag_id='Airbyte-Dag',
         default_args={'owner': 'Rajesh'},
         schedule_interval='@daily',
         start_date=days_ago(1)
    ) as dag:

    trigger_sync= AirbyteTriggerSyncOperator(
        task_id='airbyte_money_json_example',
        airbyte_conn_id='airbyte_conn',  # this is the ID we created on Airflow
        connection_id='Your Connection ID', # Get this ID by following Step 2
        asynchronous=True,
    )

    monitor_sync = AirbyteJobSensor(
        task_id='airbyte_sensor_money_json_example',
        airbyte_conn_id='airbyte_conn', # this is the ID we created on Airflow
        airbyte_job_id= trigger_sync.output
    )
    
    trigger_sync >> monitor_sync

• airbyte_conn_id: Name of the Airflow HTTP Connection pointing at the Airbyte API. Tells Airflow where the Airbyte API is located. (airbyte_conn)
• connection_id: The ID of the Airbyte Connection to be triggered by Airflow. ( ID which we copied from the Airbyte UI in Step 2 )
• asynchronous: Determines how the Airbyte Operator executes. When true, Airflow will monitor the Airbyte Job using an AirbyteJobSensor. Default value is false. ( I want my Airbyte sensor to monitor the sync process. So set asynchronous = True )

In conclusion, connecting Airflow and Airbyte to create an automated data pipeline can help you assure seamless data transfer. You can easily manage your data pipeline and make sure that your data is delivered reliably and effectively with Airflow’s robust scheduling and monitoring features. You may orchestrate your Airbyte connection with Airflow and optimize your data pipeline by following the instructions in this article. So, utilizing Airflow to coordinate your Airbyte connection will help you save time and effort while maintaining the dependability of your data pipeline, regardless of whether you’re working on a small project or a large-scale enterprise-level data pipeline. Thank you!