# RudderStack Airflow Integration


[Apache Airflow](https://airflow.apache.org/) is an open source platform to schedule, manage, and monitor workflows for data engineering pipelines.

RudderStack provides an [Airflow operator](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/index.html) for triggering your Profiles runs and Reverse ETL syncs programmatically via Airflow.

See the [GitHub Repository](https://github.com/rudderlabs/rudder-airflow-provider) for more information on the codebase with [sample DAGs](https://github.com/rudderlabs/rudder-airflow-provider/tree/main/examples).

## Prerequisites

- A working Apache Airflow setup in place. See the [Airflow documentation](https://airflow.apache.org/docs/apache-airflow/stable/installation/index.html) for more information.
- Generate a [workspace-level Service Access Token]({{< ref "access-management/service-access-tokens.md#workspace-sat" >}}) in the RudderStack dashboard with the following [permissions]({{< ref "access-management/policies-overview.md#resource-permissions" >}}):

| Resource | Permissions |
| :----| :-----|
| Tables / SQL Models / Audiences | **Edit**, **Connect** |
| Destinations | **Edit**, **Connect** | 
| Profiles | **Edit**, **Connect** |

- **For testing or development purposes only**: Generate a [Personal Access Token]({{< ref "access-management/personal-access-tokens.md" >}}) with **Read-Write** role

{{< warning >}}
**RudderStack recommends using a workspace-level Service Access Token for authentication.**

Any action authenticated by a Personal Access Token will break if the user is removed from the organization or a breaking change is made to their permissions.
{{< /warning >}}

#### Token permissions for legacy RBAC system

If you are on the [legacy Permissions Management (RBAC) system]({{< ref "archive/dashboard-guides/user-management.md" >}}), your workspace-level Service Access Token should have minimum **Admin** permissions.

See [this documentation]({{< ref "archive/dashboard-guides/service-access-tokens.md#generate-service-access-token" >}}) for more information on generating the token.

{{< image src="images/access-management/permissions/legacy/admin.webp" alt="workspace-level Service Access Token with Admin permission" >}}

## Run Airflow

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

```bash
airflow standalone
```

{{< warning >}}
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.
{{< /warning >}}

## Install Airflow operator

Install the RudderStack Airflow Provider by running the following command:

```bash
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**:

{{< image src="images/warehouse-actions-sources/airflow-provider-1.webp" alt="Airflow dashboard Connections option" >}}

2. Add a new connection by configuring the below settings:

{{< image src="images/warehouse-actions-sources/airflow-provider-2.webp" alt="Airflow dashboard edit connection" >}}

| Setting  | Description |
| :----| :-----|
| **Connection ID** |  Specify a unique connection name. By default, `RudderstackRETLOperator` / `RudderstackProfilesOperator` uses the connection name `rudderstack_default`. <br /><br />**Note**: If you have created a connection with a different name, make sure that name is passed as a parameter to the above operators. |
| **Connection Type** | Set this to **HTTP**. |
| **Host** | Set the value depending on your region.<br /><br /><ul><li>**Standard (US)**: `https://api.rudderstack.com`</li><li>**EU**: `https://api.eu.rudderstack.com`</li></ul>|
| **Password** | Enter your [workspace-level Service Access Token](#prerequisites). | 

## RudderStack operators

Use the below operators to create DAGs and schedule them via Airflow.

{{< warning >}}
Make sure to set the [schedule type]({{< ref "data-pipelines/reverse-etl/developer-guides/sync-schedule-settings.md" >}}) to **Manual** in the RudderStack dashboard for both Reverse ETL syncs and Profiles jobs.

This way, the sync jobs can be triggered only via Airflow.
{{< /warning >}}

### RudderStackRETLOperator

Use `RudderStackRETLOperator` to schedule and trigger your Reverse ETL syncs via Airflow.

A simple DAG for triggering syncs for a [Reverse ETL source]({{< ref "sources/reverse-etl/" >}}) is shown below. See this [example](https://github.com/rudderlabs/rudder-airflow-provider/tree/main/examples) for the complete code.

```python
with DAG(
    "rudderstack-retl-sample",
    default_args=default_args,
    description="A simple tutorial DAG for Reverse ETL",
    schedule_interval=timedelta(days=1),
    start_date=datetime(2021, 1, 1),
    catchup=False,
    tags=["rs-retl"],
) as dag:
    # retl_connection_id, sync_type are template fields
    rs_operator = RudderstackRETLOperator(
        retl_connection_id="<retl_connection_id>",
        task_id="<airflow_task_id>",
        connection_id="<connection_id>"
    )
```

The `RudderstackRETLOperator` parameters are described below:

| Parameter | Description | Type | Default value |
| :---|:---|:---|:---|
| `retl_connection_id`  <br/> <span style="color: #4D4DFF;font-size:12px;">Required</span> | Your [Reverse ETL Connection ID](#faq) from the RudderStack dashboard. | String (templatable) | - |
| `connection_id`  <br/> <span style="color: #4D4DFF;font-size:12px;">Required</span> | **Connection ID** used while [setting up the Airflow connection](#create-airflow-connection).  | String | `rudderstack_default` | 
| `task_id` <br/> <span style="color: #4D4DFF;font-size:12px;">Required</span> | Unique, meaningful ID for the job. See the [Airflow documentation](https://airflow.apache.org/docs/apache-airflow/1.10.9/_api/airflow/operators/index.html#package-contents) for more information on this parameter. | String |  - |
| `poll_interval` | Time (in seconds) for the polling status of triggered job. | Float | `10` |
| `poll_timeout` | Time (in seconds) after which the polling for a triggered job is declared timed out. | Float | None, that is, the provider keeps polling till the job completes or fails. |
| `sync_type` | Sync type. Acceptable values are `full` and `incremental`. | String | `None` as RudderStack determines the sync type. | 
| `request_max_retries` | Maximum number of times requests to the RudderStack API should be retried before failing. |   Integer | `3` |
| `request_retry_delay` | Time (in seconds) to wait between each request retry. | Integer | `1` |
| `request_timeout` | Time (in seconds) after which the requests to RudderStack are declared timed out. | Integer | `30` |
| `wait_for_completion` | Determines if the execution run should poll and wait till sync completion. | Boolean | `True` | 

{{< info >}}
RudderStack recommends retaining the default values for the non-mandatory parameters.
{{< /info >}}

### RudderStackProfilesOperator

Use `RudderstackProfilesOperator` to trigger a Profiles run. 

A simple DAG for triggering a Profiles project run is shown:

```python
with DAG(
    "rudderstack-profiles-sample",
    default_args=default_args,
    description="A simple tutorial DAG for Profiles run.",
    schedule_interval=timedelta(days=1),
    start_date=datetime(2021, 1, 1),
    catchup=False,
    tags=["rs-profiles"],
) as dag:
    # profile_id is template field
    rs_operator = RudderstackProfilesOperator(
        profile_id="<profiles_id>",
        task_id="<airflow_task_id>",
        connection_id="<connection_id>",
    )
```

The `RudderstackRETLOperator` parameters are described below:

| Parameter | Description | Type | Default value |
| :---|:---|:---|:---|
| `profiles_id`  <br/> <span style="color: #4D4DFF;font-size:12px;">Required</span> | Your [Profiles project ID](#faq) from the RudderStack dashboard. | String (templatable) | - |
| `connection_id`  <br/> <span style="color: #4D4DFF;font-size:12px;">Required</span> | **Connection ID** used while [setting up the Airflow connection](#create-airflow-connection).  | String | `rudderstack_default` | 
| `task_id` <br/> <span style="color: #4D4DFF;font-size:12px;">Required</span> | Unique, meaningful ID for the job. See the [Airflow documentation](https://airflow.apache.org/docs/apache-airflow/1.10.9/_api/airflow/operators/index.html#package-contents) for more information on this parameter. | String |  - |
| `parameters` | Additional parameters to pass to the Profiles run command, as supported by the API endpoint. | String | `None` |
| `poll_interval` | Time (in seconds) for the polling status of triggered job. | Float | `10` |
| `poll_timeout` | Time (in seconds) after which the polling for a triggered job is declared timed out. | Float | None, that is, the provider keeps polling till the job completes or fails. |
| `request_max_retries` | Maximum number of times requests to the RudderStack API should be retried before failing. |   Integer | `3` |
| `request_retry_delay` | Time (in seconds) to wait between each request retry. | Integer | `1` |
| `request_timeout` | Time (in seconds) after which the requests to RudderStack are declared timed out. | Integer | `30` |
| `wait_for_completion` | Determines if the execution run should poll and wait till sync completion. | Boolean | `True` | 

{{< info >}}
RudderStack recommends retaining the default values for the non-mandatory parameters.
{{< /info >}}

### 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:

```bash
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:

{{< image src="images/warehouse-actions-sources/airflow-provider-3.webp" alt="enabling Airflow DAG in dashboard" >}}

You can trigger a DAG by clicking on the play button on the right as seen above and selecting **Trigger DAG**. 

{{< warning >}}
Stopping the DAG will **not** cancel the ongoing sync.
{{< /warning >}}

## Sample DAG

A sample DAG for a Profiles run followed by a Reverse ETL sync is shown:

```python
from datetime import timedelta

from airflow import DAG

from rudder_airflow_provider.operators.rudderstack import (
	RudderstackRETLOperator,
	RudderstackProfilesOperator
)

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

with DAG('rudderstack-profiles-then-retl-sample',
    default_args=default_args,
    description='A simple tutorial DAG for Profiles run and then Reverse ETL sync.',
    catchup=False,
    tags=['rs']) as dag:
    # profile_id is a template field.
    profiles_task = RudderstackProfilesOperator(
        profile_id="{{ var.value.profile_id }}",
        task_id="<airflow_task_id>",
        connection_id="<connection_id>",
    )
    # retl_connection_id is a template field.
    retl_sync_1_task = RudderstackRETLOperator(
        retl_connection_id="{{ var.value.retl_connection_id1 }}",
        connection_id='<connection_id>',
        wait_for_completion=True,
        retry_delay=timedelta(seconds=5),
        retries=1,
    )
	# another retl sync
    retl_sync_2_task = RudderstackRETLOperator(
        retl_connection_id="{{ var.value.retl_connection_id2 }}",
        task_id="<airflow_task_id>",
        connection_id='<connection_id>'',
        wait_for_completion=True
    )
	
	# run profiles_task, then retl_sync_1_task and retl_sync_2_task in parallel
	profiles_task >> [retl_sync_1_task, retl_sync_2_task]
	
```

## 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.

{{< image src="images/retl-sources/connection-id.webp" alt="connection ID for Reverse ETL" >}}

#### Where can I find my Profiles project ID?

To obtain the Profiles project ID, go to your project in the RudderStack dashboard and note it down from the URL:

{{< image src="images/api/source-id.webp" >}}