Schedule and Trigger Reverse ETL Syncs with Airflow Provider

RudderStack’s Airflow Provider lets you programmatically schedule and trigger your Reverse ETL syncs from outside RudderStack and integrate them with your existing Airflow workflows.

For more information on the codebase and sample implementation, see the GitHub Repository.

Prerequisites

To use the Airflow Provider, you must have a working Apache Airflow installation. For more information, see the Airflow documentation.

Follow the steps in the below sections to use the RudderStack Airflow Provider:

Run Airflow

Initialize all dependencies by running Apache Airflow via the following command:

airflow standalone

The Airflow standalone server is not meant for use in production. It is highly recommended using alternate methods to install and run Airflow in a production environment.

Install Airflow Provider

Install the RudderStack Airflow Provider by running the following command:

pip install rudderstack-airflow-provider

Create Airflow connection

To create a new Airflow connection, follow these steps:

1. In your Airflow dashboard, go to Admin > Connections:

2. Add a new connection by configuring the following details:

• Connection ID: Specify a unique connection name. RudderstackRETLOperator will pick the connection with the name rudderstack_default by default. If you have created a connection with a different name, make sure that name is passed as a parameter to RudderstackRETLOperator.
• Connection Type: For this field, select HTTP from the list.
• Host: Set the value for this field depending on your region:

https://api.rudderstack.com

https://api.eu.rudderstack.com

• Password: Enter your Personal Access Token.

Define a DAG

Next, define a DAG with the tasks as per your requirement.

The following code snippet highlights an Airflow DAG with one task named rs_trigger_sync for the Reverse ETL connection ID 20dQV6yuUDUw31peWA8f7xxgHdN:

For more information on obtaining the connection ID, see the FAQ section below:

from datetime import datetime, timedelta

from airflow import DAG
from rudder_airflow_provider.operators.rudderstack import RudderstackRETLOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

with DAG(
    'rudderstack-sample',
    default_args=default_args,
    description='A simple tutorial DAG',
    schedule_interval=timedelta(days=1),
    start_date=datetime(2021, 1, 1),
    catchup=False,
    tags=['rs']
) as dag:
    rs_operator = RudderstackRETLOperator(
        retl_connection_id='2aiDQzMqP6LNuUokWstmaubcZOP',
        task_id='rs_trigger_sync',
        connection_id='rudderstack_default',
        sync_type='incremental',
        wait_for_completion=True
    )

from datetime import datetime, timedelta

from airflow import DAG
from rudder_airflow_provider.operators.rudderstack import RudderstackOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email': ['airflow@example.com'],
    'email_on_failure': False
}

with DAG(
    'rudderstack-sample',
    default_args=default_args,
    description='A simple tutorial DAG',
    schedule_interval=timedelta(days=1),
    start_date=datetime(2021, 1, 1),
    catchup=False,
    tags=['rs']
) as dag:
    rs_operator = RudderstackOperator(
        source_id='20dQV6yuUDUw31peWA8f7xxgHdN',
        connection_id='rudderstack_sync_conn',
        task_id='rs_trigger_sync',
        wait_for_completion='False'
    )

Run a DAG

Once you have defined a DAG and configured an Airflow connection, run the following commands to allow Airflow to pick up and run the DAG:

export AIRFLOW_HOME=</path/to/airflow_home>
mkdir $AIRFLOW_HOME/dags
cp rudderstack_dag.py $AIRFLOW_HOME/dags

Make sure the Airflow scheduler is running in the background. Also, you must enable the DAG in the Airflow dashboard:

You can trigger a DAG by clicking on the play button on the right as seen above, and selecting Trigger DAG. Note that stopping the DAG will not cancel the ongoing sync.

FAQ

Where can I find the connection ID for my Reverse ETL connection?

The connection ID is a unique identifier for any Reverse ETL connection set up in RudderStack.

To obtain the connection ID, click the destination connected to your Reverse ETL source and go to the Settings tab.

Questions? Contact us by email or on Slack