Event-Driven dbt with Fivetran and Prefect

Fivetran syncs can take widely varying amounts of time to complete. A connector that usually finishes in five minutes can balloon to 40 when volume spikes or the network gets flaky. If you schedule your dbt transformations on a fixed cron, you're either padding in dead time you don't need or kicking off transformations against data that hasn't fully landed.

At Prefect, we sync data from our source systems into the warehouse with Fivetran, then transform it with dbt. Instead of guessing when Fivetran will finish, we start the dbt run the moment it does. Fivetran fires a webhook when a sync completes. Prefect turns that webhook into an event, and an automation kicks off the deployment. You can even pass the connector's sync ID through as a selector, so you only rebuild the models that depend on the connector that just finished.

This tutorial wires that up end to end: a Prefect Events webhook to receive the signal, a Fivetran webhook to send it, and an automation to connect the two to your dbt deployment.

Prerequisites

• Prefect Cloud Starter Tier or above
• A Fivetran API key
• A Prefect deployment configured to accept parameters

If you don't have a Prefect deployment yet, you can deploy this quickstart to Prefect's infrastructure. If you already have a dbt project, use the prefect-dbt integration to turn it into a Prefect flow and deploy it to Prefect's managed infrastructure.

Step 1: Create a Prefect Events webhook

Fivetran needs somewhere to send its sync events, and Prefect needs to know how to read them. A Prefect Events webhook handles both: it gives you a URL to point Fivetran at, and a template that turns the incoming payload into a Prefect event.

• Navigate to your Prefect Cloud workspace
• Go to Settings → Webhooks in the left sidebar
• Click the + button at the top of the page

Creating a new Prefect Events webhook in the Prefect Cloud UI

Configure the webhook:

• Name: fivetran-sync-events
• Description: Receives Fivetran sync completion events
• Template: Dynamic
• Template body:

{
  "event": "{{ body.event }}",
  "resource": {
    "prefect.resource.id": "fivetran_connector_id.{{ body.connector_id }}",
    "producing-team": "Data Engineering"
  }
}

You don't have to use this exact JSON, but this version creates an event in Prefect called "Sync end" and carries the connector ID through on the resource, which is what lets you target a specific connector in an automation later.

Click Create, then copy the generated webhook URL (for example, https://api.prefect.cloud/hooks/YOUR_PREFECT_WEBHOOK). You'll hand it to Fivetran in the next step.

Step 2: Create a Fivetran webhook

Fivetran webhooks can currently only be created through the API, so the cleanest approach is a small Python script you can run to create, test, and tear down the webhook. Drawing on the samples in the Fivetran webhook API docs, the script below does all three:

import requests
 
 
def create(webhook_url):
    """Create a Fivetran webhook that sends events to Prefect Cloud"""
    url = "https://api.fivetran.com/v1/webhooks/account"
    payload = {
        "url": webhook_url,
        "events": ["sync_end"],
        "active": True,
    }
    headers = {
        "Accept": "application/json",
        "Authorization": "Basic YOUR_BASE64_ENCODED_API_KEY",  # Replace with your key
        "content-type": "application/json",
    }
    response = requests.request("POST", url, json=payload, headers=headers)
    response.raise_for_status()
    response_data = response.json()
    print(response.text)
    if response_data.get("code") == "Success" and "data" in response_data:
        return response_data["data"]["id"]
 
 
def test(webhook_id):
    """Test the webhook to ensure it's working"""
    url = f"https://api.fivetran.com/v1/webhooks/{webhook_id}/test"
    payload = {"event": "sync_end"}
    headers = {
        "Accept": "application/json",
        "Authorization": "Basic YOUR_BASE64_ENCODED_API_KEY",  # Replace with your key
        "content-type": "application/json",
    }
    response = requests.request("POST", url, json=payload, headers=headers)
    response.raise_for_status()
    print(response.text)
 
 
def delete(webhook_id):
    """Delete a webhook when no longer needed"""
    url = f"https://api.fivetran.com/v1/webhooks/{webhook_id}"
    headers = {
        "Accept": "application/json",
        "Authorization": "Basic YOUR_BASE64_ENCODED_API_KEY",  # Replace with your key
    }
    response = requests.request("DELETE", url, headers=headers)
    print(response.text)
 
 
if __name__ == "__main__":
    webhook_id = create("https://api.prefect.cloud/hooks/YOUR_PREFECT_WEBHOOK")
    if webhook_id:
        test(webhook_id)
    # delete(webhook_id)

Running this creates the webhook and immediately fires a test sync_end event at your Prefect URL. Head to the Events page in Prefect Cloud and you should see a new "Sync end" event waiting for you.

The Sync end event appearing on the Prefect Cloud Events page

Step 3: Trigger your deployment from the event

The event is arriving; now you need something to listen for it. An automation watches for "Sync end" events and runs your deployment when one shows up.

Click into the "Sync end" event, open the kebab menu in the top right of the event page, and choose Automate. Prefect pre-populates the trigger section of a new automation from the event you're looking at, so you don't have to hand-write the matching criteria.

Automating directly from the Sync end event, with the trigger pre-populated

Configure the action:

• Action Type: Run a Deployment
• Deployment: Select your deployment
• Parameters: Open the kebab menu next to Parameters, choose JSON, and pull the connector ID out of the event

Configuring the Run a Deployment action and passing parameters as JSON

{
  "YOUR_PARAMETER_NAME": "{{ event.resource['prefect.resource.id'] | replace('fivetran_connector_id.', '') }}"
}

This strips the fivetran_connector_id. prefix back off and passes the bare connector ID into your deployment as YOUR_PARAMETER_NAME. To confirm the whole chain works, re-run test(webhook_id) from the script in Step 2 with the webhook ID returned by create(), and watch the automation fire a deployment run.

Use that parameter in a dbt flow as a selector, and dbt rebuilds its models the instant the upstream Fivetran sync finishes. This minimizes latency and reruns.

If your deployment uses the prefect-dbt integration, the Assets tab in the Prefect UI gives you the view across it all:

dbt assets and their dependencies shown in the Prefect Assets tab

Materialization history for dbt models in the Assets tab

Assets let you see every dbt model and its dependencies, when each one was last materialized, and a general read on how healthy your data is.

Next steps

If you sync from a variety of source systems, you don't have to wait for every connector to finish before triggering one big dbt job. When a Stripe sync completes, for instance, you can hand its connector ID to a dbt selector that builds only the models that connector touches. Stakeholders get fresher dashboards without waiting on your slowest sync to catch up.

To go further with Prefect:

• Visit prefect.io
• Explore Prefect on GitHub to open issues and pull requests
• Follow Prefect on LinkedIn for regular updates
• Join our active Slack community