SyncSpur | Orchestrating Fivetran Syncs with Airflow and Dagster (API, Sensors and Webhooks)

Fivetran's scheduler is deliberately simple: each connection syncs on its own interval, and that is the whole model. It works beautifully until the day someone asks a question the scheduler cannot answer — "why did the finance dashboard show yesterday's revenue against today's orders?" The answer is almost always the same. Salesforce finished at 09:04, NetSuite finished at 09:19, and the transformation job ran at 09:10 on a cron because that felt about right.

Time-based coupling is the bug. The fix is to let an orchestrator observe pipeline state rather than guess at pipeline timing. This tutorial shows three ways to do that with Fivetran — the REST API, webhooks, and native integrations — and, just as importantly, when not to bother.

Decide first: do you actually need an orchestrator?

Before writing any DAG code, check whether one of the built-in options already solves it:

Fivetran Transformations (Quickstart models or a dbt project run by Fivetran) can be set to run on connection completion. If your only requirement is "run dbt after these connectors land", this is free, native, and needs no orchestrator at all.
Connection scheduling with daily_sync_time gets you a predictable landing window without any external code.

Reach for Airflow or Dagster when you have real cross-system dependencies: Fivetran connections plus a Spark job plus an external API extract plus a reverse-ETL push, all with retry, alerting and backfill semantics that must live in one place. That is a legitimate need. "We already have Airflow" is not, on its own, a good enough reason to put a scheduler in front of another scheduler.

The API surface you will use

Everything below runs against the Fivetran REST API v1 with an account-level API key and secret, sent as HTTP Basic auth. Three endpoints carry almost all the weight:

Endpoint	Purpose
`POST /v1/connections/{id}/sync`	Trigger a sync. Body `{"force": true}` interrupts a running sync; default queues behind it.
`GET /v1/connections/{id}`	Read `status.sync_state`, `succeeded_at`, `failed_at`, `status.setup_state`.
`POST /v1/connections/{id}/resync`	Full historical re-sync. Handle with extreme care — see the warning below.

A minimal poll-until-done helper in Python:

import time, requests
from requests.auth import HTTPBasicAuth

BASE = "https://api.fivetran.com/v1"

def sync_and_wait(conn_id, key, secret, timeout=7200, interval=30):
    auth = HTTPBasicAuth(key, secret)
    before = requests.get(f"{BASE}/connections/{conn_id}", auth=auth).json()
    prev_succeeded = before["data"].get("succeeded_at")

    requests.post(f"{BASE}/connections/{conn_id}/sync",
                  auth=auth, json={"force": False}).raise_for_status()

    deadline = time.time() + timeout
    while time.time() < deadline:
        data = requests.get(f"{BASE}/connections/{conn_id}", auth=auth).json()["data"]
        state = data["status"]["sync_state"]          # scheduled | syncing | paused | rescheduled
        if state == "scheduled":
            if data.get("succeeded_at") != prev_succeeded:
                return "success"
            if data.get("failed_at") and data["failed_at"] > (prev_succeeded or ""):
                raise RuntimeError(f"sync failed for {conn_id}")
        time.sleep(interval)
    raise TimeoutError(f"{conn_id} still syncing after {timeout}s")

Two details that trip people up. First, sync_state returns to scheduled for both success and failure, so you must compare succeeded_at and failed_at timestamps against what they were before you triggered — the state alone tells you nothing about the outcome. Second, capture prev_succeeded before the trigger, not after; otherwise a fast connector can complete between your trigger and your first poll and you will wait out the full timeout on a sync that already worked.

Airflow: use the provider, not a PythonOperator

apache-airflow-providers-fivetran gives you an operator and a deferrable sensor. The sensor is the part that matters: a reschedule-mode or deferred sensor frees the worker slot while Fivetran does its work, instead of burning a slot for forty minutes on a time.sleep loop.

from airflow.decorators import dag
from airflow.providers.fivetran.operators.fivetran import FivetranOperator
from airflow.providers.fivetran.sensors.fivetran import FivetranSensor
from pendulum import datetime

@dag(start_date=datetime(2026, 1, 1), schedule="0 6 * * *", catchup=False)
def revenue_pipeline():
    trigger = FivetranOperator(
        task_id="sync_salesforce",
        fivetran_conn_id="fivetran_default",
        connector_id="{{ var.value.sf_connector_id }}",
        wait_for_completion=False,
    )
    wait = FivetranSensor(
        task_id="wait_salesforce",
        fivetran_conn_id="fivetran_default",
        connector_id="{{ var.value.sf_connector_id }}",
        poke_interval=60,
        deferrable=True,
    )
    trigger >> wait  # >> dbt_run >> reverse_etl_push

revenue_pipeline()

Set the DAG-level max_active_runs=1 on anything that triggers syncs. Overlapping runs firing force: true at the same connection is a reliable way to generate half-finished loads and a surprising MAR bill.

Dagster: model the tables as assets

Dagster's dagster-fivetran integration takes a different and, for analytics work, generally better angle. load_assets_from_fivetran_instance (or build_fivetran_assets_definitions) turns each synced table into a software-defined asset, so your dbt models can declare an upstream dependency on salesforce/opportunity rather than on a task named sync_salesforce.

from dagster import Definitions, EnvVar
from dagster_fivetran import FivetranWorkspace, build_fivetran_assets_definitions

fivetran = FivetranWorkspace(
    account_id=EnvVar("FIVETRAN_ACCOUNT_ID"),
    api_key=EnvVar("FIVETRAN_API_KEY"),
    api_secret=EnvVar("FIVETRAN_API_SECRET"),
)

defs = Definitions(
    assets=build_fivetran_assets_definitions(workspace=fivetran),
    resources={"fivetran": fivetran},
)

The payoff is lineage: one graph from source connector through warehouse tables to dbt models to the reverse-ETL sync, with freshness policies attached. When the CFO asks why a number is stale, you point at the asset that has not materialised instead of reading three log files.

Webhooks: event-driven instead of polling

Polling costs you API calls and latency. If your orchestrator can accept an inbound HTTP call, subscribe to sync_end instead:

curl -X POST https://api.fivetran.com/v1/webhooks/account \
  -u "$FIVETRAN_API_KEY:$FIVETRAN_API_SECRET" \
  -H "Content-Type: application/json" \
  -d '{
        "url": "https://orchestrator.example.com/hooks/fivetran",
        "events": ["sync_end", "sync_start"],
        "active": true,
        "secret": "'"$WEBHOOK_SHARED_SECRET"'"
      }'

Fivetran signs each delivery with HMAC-SHA256 in the X-Fivetran-Signature-256 header; verify it against your shared secret with a constant-time comparison and reject anything that fails. The sync_end payload includes the connection ID and a status of SUCCESSFUL, FAILURE, RESCHEDULED or FAILURE_WITH_TASK — only the first should advance your DAG. In Airflow this maps cleanly onto a TriggerDagRunRequest against the stable REST API; in Dagster, onto a sensor that reports a RunRequest.

Webhooks are at-least-once. Make the receiving endpoint idempotent, keyed on connection ID plus the sync completion timestamp, or a retried delivery will kick off a duplicate transformation run.

The one endpoint to guard

POST /v1/connections/{id}/resync re-reads history from the source and re-lands it. On a connection-level MAR plan that can turn a routine deploy into a five-figure invoice, and on a large ERP source it can take days. Never expose it in a scheduled DAG. If you need it in code, keep it behind a manually-triggered job with an explicit confirmation parameter, and require a second approver. We have been called in to clean up after an automated retry loop that called resync on failure; the pipeline recovered, the budget did not.

A workable pattern

For most teams the sequence that holds up in production looks like this:

Leave connections on their native Fivetran schedule — do not trigger them from the orchestrator unless you genuinely need to control timing.
Have the orchestrator wait on completion (webhook preferred, deferrable sensor otherwise) rather than trigger.
Fan into transformations only once every required connection reports success for the current window.
Alert on the gap between expected and actual completion, using the Fivetran Platform Connector for the historical baseline.
Keep force: true and resync out of automation entirely.

That gives you correct dependencies without duplicating Fivetran's scheduler, and it keeps the failure modes boring.

Where this goes wrong at scale

Three failure patterns show up repeatedly in orchestration reviews. Paused connections that a sensor waits on forever, because sync_state: paused is not an error and most naive loops do not check for it — test for it explicitly and fail fast. Time zones, where the orchestrator schedules in UTC and daily_sync_time was set in local time, producing a one-hour dependency gap twice a year. And rate limiting: the API allows a limited number of calls per minute per account, so a 15-second poll interval across sixty connections will start returning 429s. Poll at 60 seconds or move to webhooks.

If your Fivetran deployment has outgrown per-connector scheduling and you want the dependency graph designed properly — or you have an Airflow DAG that is quietly re-syncing something expensive — get in touch. Orchestration design, cost review and pipeline rescue are core parts of what we do.