Reverse ETL: Data Terug naar je Businesstools

Wat is Reverse ETL?

In de klassieke data-architectuur stroomt data altijd in één richting: vanuit bronsystemen (CRM, ERP, webshop) via een ETL-pipeline naar een centraal datawarehouse of data lakehouse. Daar wordt de data opgeslagen, getransformeerd en geanalyseerd. Maar dan? Dan opent een analist een dashboard in Looker of Power BI, trekt conclusies… en gaat de data nergens anders meer naartoe.

Reverse ETL keert dit proces om. In plaats van data naar het warehouse te trekken, stuur je de verrijkte, getransformeerde data vanuit je warehouse terug naar de operationele tools waar je team dagelijks mee werkt: je CRM, marketing automation platform, customer success tool, of zelfs je eigen applicatie.

Waarom is Reverse ETL in 2026 relevanter dan ooit?

De moderne data stack heeft zich de afgelopen jaren radicaal ontwikkeld. Teams investeren fors in datawarehouses (Snowflake, BigQuery, Databricks), in data transformatietools (dbt) en in observability. Het resultaat? Rijke, goed gedocumenteerde datasets met gesegmenteerde klanten, voorspellingsmodellen en 360°-klantprofielen. Maar al die rijkdom blijft opgesloten in het warehouse — onbereikbaar voor de salesrep die zijn pipeline beheert in Salesforce, of de marketeer die campagnes opbouwt in HubSpot.

In 2026 zijn er een aantal trends die Reverse ETL tot een must-have maken:

Hoe werkt Reverse ETL? Architectuur en Dataflow

Laten we de architectuur stap voor stap doorlopen. Een typische Reverse ETL-setup bestaat uit vier componenten: het warehouse, een transformatielaag (dbt), een Reverse ETL-tool en de doelbestemming.

Dataflow Overzicht

Praktische Codevoorbeelden

Laten we concreet worden. We werken hier met een realistisch scenario: een SaaS-bedrijf wil klanten met een hoog churn-risico identificeren en deze segmentatie synchroniseren naar HubSpot als een custom property, zodat het sales team direct kan handelen.

Stap 1: dbt-model voor klantprofielen (BigQuery)

We bouwen eerst een dbt-model dat klantprofielen aanmaakt inclusief churn-risicoscore:

-- models/marts/customer_profiles.sql
-- dbt-model: customer_profiles

WITH base_customers AS (
    SELECT
        customer_id,
        email,
        company_name,
        plan_type,
        mrr,
        created_at
    FROM {{ ref('stg_customers') }}
),

usage_metrics AS (
    SELECT
        customer_id,
        COUNT(DISTINCT DATE(event_date))         AS active_days_last_30,
        COUNT(*)                                  AS total_events_last_30,
        MAX(event_date)                           AS last_active_date,
        COUNTIF(event_type = 'feature_advanced')  AS advanced_feature_usage
    FROM {{ ref('stg_events') }}
    WHERE event_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
    GROUP BY 1
),

support_tickets AS (
    SELECT
        customer_id,
        COUNT(*) AS open_tickets_count
    FROM {{ ref('stg_support_tickets') }}
    WHERE status = 'open'
    GROUP BY 1
),

churn_scoring AS (
    SELECT
        c.customer_id,
        c.email,
        c.company_name,
        c.plan_type,
        c.mrr,
        COALESCE(u.active_days_last_30, 0)        AS active_days_last_30,
        COALESCE(u.last_active_date, c.created_at) AS last_active_date,
        COALESCE(u.advanced_feature_usage, 0)      AS advanced_feature_usage,
        COALESCE(s.open_tickets_count, 0)          AS open_tickets_count,
        -- Eenvoudige rule-based churn score (0-100)
        CASE
            WHEN COALESCE(u.active_days_last_30, 0) = 0 THEN 90
            WHEN COALESCE(u.active_days_last_30, 0) < 5 THEN 70
            WHEN COALESCE(u.active_days_last_30, 0) < 10 THEN 45
            WHEN COALESCE(u.advanced_feature_usage, 0) = 0 THEN 35
            ELSE 10
        END AS churn_risk_score,
        CASE
            WHEN COALESCE(u.active_days_last_30, 0) = 0      THEN 'critical'
            WHEN COALESCE(u.active_days_last_30, 0) < 5       THEN 'high'
            WHEN COALESCE(u.active_days_last_30, 0) < 10      THEN 'medium'
            ELSE 'low'
        END AS churn_risk_segment
    FROM base_customers c
    LEFT JOIN usage_metrics u   USING (customer_id)
    LEFT JOIN support_tickets s USING (customer_id)
)

SELECT
    customer_id,
    email,
    company_name,
    plan_type,
    mrr,
    active_days_last_30,
    last_active_date,
    advanced_feature_usage,
    open_tickets_count,
    churn_risk_score,
    churn_risk_segment,
    CURRENT_TIMESTAMP() AS updated_at
FROM churn_scoring

Stap 2: Census Sync configureren via de API

Census biedt naast een UI ook een REST API waarmee je syncs programmatisch kunt aanmaken en triggeren. Hier is een Python-script dat een sync triggert na een succesvolle dbt-run:

import requests
import os
import json

CENSUS_API_KEY = os.environ["CENSUS_API_KEY"]
CENSUS_SYNC_ID = os.environ["CENSUS_SYNC_ID"]  # ID van je HubSpot sync

BASE_URL = "https://bearer.census.app/api/v1"

def trigger_census_sync(sync_id: str, full_sync: bool = False) -> dict:
    """
    Triggert een Census Reverse ETL sync via de REST API.
    
    Args:
        sync_id: Het ID van de Census sync definitie
        full_sync: Als True, synchroniseert alle records (niet alleen delta's)
    
    Returns:
        Sync run metadata inclusief run_id voor status polling
    """
    headers = {
        "Authorization": f"Bearer {CENSUS_API_KEY}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "full_sync": full_sync
    }
    
    response = requests.post(
        f"{BASE_URL}/syncs/{sync_id}/trigger",
        headers=headers,
        json=payload
    )
    response.raise_for_status()
    return response.json()


def get_sync_run_status(sync_id: str, sync_run_id: str) -> dict:
    """Haalt de huidige status op van een lopende sync run."""
    headers = {"Authorization": f"Bearer {CENSUS_API_KEY}"}
    
    response = requests.get(
        f"{BASE_URL}/syncs/{sync_id}/sync_runs/{sync_run_id}",
        headers=headers
    )
    response.raise_for_status()
    return response.json()


def wait_for_sync_completion(sync_id: str, sync_run_id: str, 
                              poll_interval: int = 30, max_wait: int = 3600) -> str:
    """
    Poll de sync status totdat deze voltooid is of een fout optreedt.
    
    Returns:
        Eindstatus: 'completed', 'failed', of 'cancelled'
    """
    import time
    
    elapsed = 0
    terminal_states = {"completed", "failed", "cancelled"}
    
    while elapsed < max_wait:
        status_data = get_sync_run_status(sync_id, sync_run_id)
        status = status_data["data"]["status"]
        records_processed = status_data["data"].get("records_processed", 0)
        
        print(f"[{elapsed}s] Status: {status} | Records verwerkt: {records_processed}")
        
        if status in terminal_states:
            return status
        
        time.sleep(poll_interval)
        elapsed += poll_interval
    
    raise TimeoutError(f"Sync niet voltooid binnen {max_wait} seconden")


if __name__ == "__main__":
    print("Census Reverse ETL sync starten voor HubSpot klantprofielen...")
    
    # Trigger de sync
    run_response = trigger_census_sync(CENSUS_SYNC_ID)
    run_id = run_response["data"]["sync_run_id"]
    print(f"Sync gestart met run_id: {run_id}")
    
    # Wacht op voltooiing
    final_status = wait_for_sync_completion(CENSUS_SYNC_ID, run_id)
    
    if final_status == "completed":
        print("✅ Sync succesvol voltooid! HubSpot is bijgewerkt.")
    else:
        print(f"❌ Sync mislukt met status: {final_status}")
        exit(1)

Stap 3: Hightouch als alternatief (YAML-configuratie)

Hightouch biedt ook een GitOps-workflow via YAML-configuratiebestanden. Zo definieer je een sync als code:

# hightouch/syncs/hubspot_customer_profiles.yml
version: "1"

name: "HubSpot - Customer Churn Profiles"
description: "Synchroniseert klantprofielen met churn scores naar HubSpot Contacts"

source:
  connection: "bigquery-production"
  model:
    type: "dbt_model"
    dbt_model_name: "customer_profiles"
  # Primaire sleutel voor change detection
  primary_key: "customer_id"

destination:
  connection: "hubspot-production"
  object: "contacts"
  # Zoek bestaande HubSpot contacten op via email
  match_on:
    hubspot_field: "email"
    model_field: "email"

# Veld-mapping: warehouse kolom → HubSpot property
field_mappings:
  - model_field: "company_name"
    destination_field: "company"
  
  - model_field: "mrr"
    destination_field: "monthly_recurring_revenue__c"
  
  - model_field: "churn_risk_score"
    destination_field: "churn_risk_score__c"
  
  - model_field: "churn_risk_segment"
    destination_field: "churn_risk_segment__c"
  
  - model_field: "active_days_last_30"
    destination_field: "active_days_last_30__c"
  
  - model_field: "last_active_date"
    destination_field: "last_login_date__c"

# Sync gedrag
behavior:
  on_match: "update"          # Update bestaande contacten
  on_no_match: "skip"         # Maak geen nieuwe contacten aan
  sync_mode: "incremental"    # Alleen gewijzigde records

# Schedule: elke 4 uur
schedule:
  type: "interval"
  interval_minutes: 240

Reverse ETL Tools: Census vs Hightouch vs Zelf Bouwen

Er zijn meerdere manieren om Reverse ETL te implementeren. De twee dominante commerciële tools zijn Census en Hightouch, maar sommige teams kiezen ervoor om zelf iets te bouwen. Hier een eerlijke vergelijking:

Best Practices en Production Tips

Reverse ETL klinkt simpel in theorie, maar in productie zijn er flink wat valkuilen. Hier zijn de belangrijkste best practices die we aanbevelen:

Aanvullende Production Tips

# Airflow DAG: dbt run gevolgd door Census sync trigger
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.dbt.cloud.operators.dbt import DbtCloudRunJobOperator
from datetime import datetime, timedelta
import requests, os

default_args = {
    "owner": "data-engineering",
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
    "email_on_failure": True,
    "email": ["data-team@bedrijf.nl"]
}

def trigger_census_sync(**context):
    """Triggert Census sync na succesvolle dbt run."""
    response = requests.post(
        f"https://bearer.census.app/api/v1/syncs/{os.environ['CENSUS_SYNC_ID']}/trigger",
        headers={"Authorization": f"Bearer {os.environ['CENSUS_API_KEY']}"},
        json={"full_sync": False}
    )
    response.raise_for_status()
    run_id = response.json()["data"]["sync_run_id"]
    print(f"Census sync gestart: run_id={run_id}")
    # Sla run_id op voor downstream monitoring
    context['task_instance'].xcom_push(key='census_run_id', value=run_id)

with DAG(
    dag_id="reverse_etl_customer_profiles",
    default_args=default_args,
    schedule_interval="0 */4 * * *",  # Elke 4 uur
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["reverse-etl", "census", "hubspot"]
) as dag:

    # Stap 1: Run het dbt transformatie-model
    dbt_run = DbtCloudRunJobOperator(
        task_id="dbt_run_customer_profiles",
        job_id=int(os.environ["DBT_CLOUD_JOB_ID"]),
        check_interval=30,
        timeout=1800
    )

    # Stap 2: Trigger de Reverse ETL sync
    census_sync = PythonOperator(
        task_id="trigger_census_sync",
        python_callable=trigger_census_sync,
        provide_context=True
    )

    # Pipeline definitie
    dbt_run >> census_sync

Data Quality Checks vóór de Sync

Stuur nooit data naar je CRM zonder minimale kwaliteitsvalidatie. Een churn score van NULL of een email met een typefout kan grote schade aanrichten in je salesproces. Voeg dbt-tests toe:

# models/marts/customer_profiles.yml
version: 2

models:
  - name: customer_profiles
    description: "Klantprofielen met churn-risicoscores voor Reverse ETL naar HubSpot"
    
    columns:
      - name: customer_id
        description: "Unieke klant-identifier (primary key voor Census)"
        tests:
          - unique
          - not_null

      - name: email
        description: "E-mailadres van de klant"
        tests:
          - not_null
          - unique

      - name: churn_risk_score
        description: "Churn risicoscore van 0 (laag) tot 100 (kritiek)"
        tests:
          - not_null
          - accepted_range:
              min_value: 0
              max_value: 100

      - name: churn_risk_segment
        description: "Churn risico segment classificatie"
        tests:
          - not_null
          - accepted_values:
              values: ['low', 'medium', 'high', 'critical']

Conclusie: Wanneer Wel en Wanneer Niet?

Reverse ETL is een krachtige toevoeging aan