Trigger syncs with Airflow

Overview

Hightouch provides an Airflow Operator for triggering syncs via Airflow. The Git repository contains the latest code as well as example DAGs to help you get started.

Setup

Create an API key

The first step is to create a Hightouch API key in your Hightouch workspace settings.

1. From the API keys tab on the Settings page, select Add API key.
2. Enter a descriptive Name for your key.
3. Copy your API key and store it in a safe location. The key will only be displayed once.
4. Click Create API key.

Add the Airflow connection

1. In Airflow, create a new connection by going to Admin > Connections and clicking the + button to add a new connection.
2. Enter hightouch_default as the Conn ID, select HTTP as the Conn Type, and set the host to https://api.hightouch.com. Optionally, enter a Description.
3. Enter the API key you created in the Password field.

Set sync schedule type

For the Airflow extension to trigger syncs via the Hightouch API, you must set the syncs' schedules to Manual.

You can do this by going to the Syncs overview page, selecting the particular sync you want to trigger with Airflow, and opening the Schedule tab. Here, make sure the Schedule type is set to Manual. You should do this for each sync you want to trigger with Airflow.

You can also find the sync ID on this page, which you need for scheduling syncs in the next step.

Install the Airflow package

While the details will vary for your particular Airflow installation, the simplest way to install our package is through pip.

pip install airflow-provider-hightouch

Add the HightouchTriggerSyncOperator to your DAG

The last step is to add the operator to a DAG and enter the sync ID of the sync you want to trigger. When the task is run, Airflow sends a call to the Hightouch API to trigger a run which completes asynchronously.

Alternatively, you can choose to send a synchronous request and await the results of the sync before completing the task.

For a full description of the operator parameters, view the source code in the GitHub repository.

from airflow_provider_hightouch.operators.hightouch \
    import HightouchTriggerSyncOperator

with DAG(....) as dag:
...
    # This task submits the request but doesn't wait for completion of the sync before completing.
    my_async_task = HightouchTriggerSyncOperator(
        task_id="run_my_sync", sync_id=1234
        )

    # This task waits for the sync status to complete or error before completing.
    # Warnings can be treated as errors or successes depending on your use case.
    my_sync_task = HightouchTriggerSyncOperator(
        task_id="run_sync", sync_id=1234, synchronous=True, error_on_warning=True
    )