Fivetran: Automated Data Movement & ELT Pipeline Guide

Fivetran is the leading fully managed ELT platform that moves data from 500+ sources into your data warehouse automatically — handling schema changes, API updates, and incremental sync with zero maintenance on your end.

What is Fivetran?

Fivetran is a fully managed ELT (Extract, Load, Transform) data movement platform. It provides pre-built, pre-maintained connectors for hundreds of data sources — SaaS applications (Salesforce, HubSpot, Stripe, Google Ads), databases (Postgres, MySQL, SQL Server, Oracle), cloud storage (S3, GCS), and event platforms (Kafka, Segment). Fivetran handles all the infrastructure and maintenance: API authentication, pagination, rate limiting, schema evolution, incremental sync logic, and error recovery.

The ELT model loads raw, normalized data into your data warehouse first — Snowflake, BigQuery, Redshift, Databricks, or others — and transformations are handled separately with dbt or SQL. This gives your team access to raw data immediately and keeps transformations auditable in version control. Fivetran adds metadata columns to every synced table: _fivetran_synced (last sync timestamp) and _fivetran_deleted (soft delete flag).

Connector Setup via API

# Fivetran Python SDK — manage connectors programmatically
# pip install fivetran-connector-sdk requests

import requests

FIVETRAN_API_KEY = "your_api_key"
FIVETRAN_API_SECRET = "your_api_secret"
BASE_URL = "https://api.fivetran.com/v1"

auth = (FIVETRAN_API_KEY, FIVETRAN_API_SECRET)

# Create a Postgres source connector
connector_payload = {
    "service": "postgres",
    "group_id": "your_destination_group_id",
    "config": {
        "host": "db.example.com",
        "port": 5432,
        "database": "production",
        "user": "fivetran_reader",
        "password": "secure_password",
        "schema": "public",
        "update_method": "WAL",            # logical replication (CDC)
        "publication_name": "fivetran_pub",
    },
}

resp = requests.post(
    f"{BASE_URL}/connectors",
    auth=auth,
    json=connector_payload,
)
connector_id = resp.json()["data"]["id"]
print(f"Connector created: {connector_id}")

# Trigger a manual sync
requests.post(f"{BASE_URL}/connectors/{connector_id}/force", auth=auth)

# Check sync status
status = requests.get(
    f"{BASE_URL}/connectors/{connector_id}",
    auth=auth
).json()["data"]["status"]
print(f"Sync status: {status['sync_state']}")

dbt Integration & Webhook Trigger

# Fivetran + dbt Cloud integration — trigger dbt after sync
# In Fivetran Dashboard: Transformations > dbt Cloud

# Or trigger via dbt Cloud API after Fivetran webhook
import requests

DBT_CLOUD_TOKEN = "your_dbt_cloud_token"
DBT_ACCOUNT_ID = "12345"
DBT_JOB_ID = "67890"

def trigger_dbt_job(event_data):
    """Called from a webhook receiver after Fivetran sync completes."""
    if event_data["event_type"] != "sync_end":
        return

    resp = requests.post(
        f"https://cloud.getdbt.com/api/v2/accounts/{DBT_ACCOUNT_ID}/jobs/{DBT_JOB_ID}/run/",
        headers={
            "Authorization": f"Token {DBT_CLOUD_TOKEN}",
            "Content-Type": "application/json",
        },
        json={"cause": f"Triggered by Fivetran sync {event_data['sync_id']}"},
    )
    run_id = resp.json()["data"]["id"]
    print(f"dbt job triggered, run_id={run_id}")

Built-in Transformations with dbt Core

# Fivetran Transformations — built-in dbt Core transformations
# Configure directly in Fivetran Dashboard > Transformations

# Fivetran runs dbt Core models after each sync.
# Example dbt model in Fivetran's transformation layer:

-- models/fivetran_staging/stg_stripe_charges.sql
WITH source AS (
    SELECT * FROM {{ source('stripe', 'charge') }}
),
cleaned AS (
    SELECT
        id                  AS charge_id,
        amount / 100.0      AS amount_usd,
        currency,
        status,
        created             AS charged_at,
        customer            AS customer_id,
        metadata:order_id   AS order_id     -- JSON field extraction
    FROM source
    WHERE _fivetran_deleted = FALSE          -- exclude soft-deleted rows
)
SELECT * FROM cleaned;

-- Fivetran adds _fivetran_synced and _fivetran_deleted columns
-- to every table automatically — always filter on _fivetran_deleted

Fivetran vs Stitch vs Airbyte vs Custom ETL