The Customer Data Platform for Developers
The RudderStack Airflow Provider lets you programmatically schedule and trigger your Reverse ETL syncs and Profiles runs outside RudderStack and integrate them with your existing Airflow workflows. Refer to orchestration docs.
pip install rudderstack-airflow-provider
Note
Use RudderstackRETLOperator for reverse ETL connections
A simple DAG for triggering syncs for a RudderStack Reverse ETL source:
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="connection_id",
task_id="<a unique, meaningful id for the airflow task>",
connection_id="<rudderstack api airflow connection id>"
)
For the complete code, refer to this example.
Mandatatory parameters for RudderstackRETLOperator:
- retl_connection_id: This is the connection id for the sync job.
- connection_id: The Airflow connection to use for connecting to the Rudderstack API. Default value is
rudderstack_default
.
RudderstackRETLOperator exposes other configurable parameters as well. Mostly default values for them would be recommended.
- request_max_retries: The maximum number of times requests to the RudderStack API should be retried before failng.
- request_retry_delay: Time (in seconds) to wait between each request retry.
- request_timeout: Time (in seconds) after which the requests to RudderStack are declared timed out.
- poll_interval: Time (in seconds) for polling status of triggered job.
- poll_timeout: Time (in seconds) after which the polling for a triggered job is declared timed out.
- wait_for_completion: Boolean if execution run should poll and wait till completion of sync. Default value is True.
- sync_type: Type of sync to trigger
incremental
orfull
. Default is None as RudderStack will be deteriming sync type.
RudderstackProfilesOperator can be used to trigger profiles run. A simple DAG for triggering profile runs for a profiles project.
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="<profile_id>",
task_id="<a unique, meaningful id for the airflow task",
connection_id="<rudderstack api connection id>",
)
Mandatatory parameters for RudderstackProfilesOperator:
- profile_id: This is the profiles id for the profiles project to run.
- connection_id: The Airflow connection to use for connecting to the Rudderstack API. Default value is
rudderstack_default
.
RudderstackProfilesOperator exposes other configurable parameters as well. Mostly default values for them would be recommended.
- request_max_retries: The maximum number of times requests to the RudderStack API should be retried before failng.
- request_retry_delay: Time (in seconds) to wait between each request retry.
- request_timeout: Time (in seconds) after which the requests to RudderStack are declared timed out.
- poll_interval: Time (in seconds) for polling status of triggered job.
- poll_timeout: Time (in seconds) after which the polling for a triggered job is declared timed out.
- wait_for_completion: Boolean if execution run should poll and wait till completion of sync. Default value is True.
- parameters: Additional parameters to pass to the profiles run command, as supported by the API endpoint. Default value is
None
.
We would love to see you contribute to this project. Get more information on how to contribute here.
The RudderStack Airflow Provider is released under the MIT License.
For more information or queries on this feature, you can contact us or start a conversation in our Slack community.