Data Reverse Etl

Reverse ETL

Purpose

Sync aggregated, transformed data from a data warehouse to operational SaaS tools (CRM, marketing, support) to power customer-facing and internal operational workflows.

Agent Protocol

Trigger

Exact user phrases: "reverse ETL", "Census", "Hightouch", "Grouparoo", "operational analytics", "warehouse to SaaS", "data activation", "warehouse sync", "audience export", "sync config", "operational data", "warehouse-first", "sync identity resolution".

Input Context

Before activating, verify:

• Source warehouse (Snowflake, BigQuery, Redshift, Databricks, Postgres)
• Destination SaaS tools (Salesforce, HubSpot, Marketo, Braze, Amplitude, Zendesk, Google Ads, Facebook Audiences)
• Sync frequency (real-time, hourly, daily, batch window)
• Identity resolution strategy (unique keys, merge rules, foreign key mapping)
• Volume (rows per sync, daily active records)
• Idempotency requirements (upsert vs replace vs append)

Output Artifact

Reverse ETL pipeline config with SQL source query, sync schedule, identity mapping, and destination-specific operation config as YAML and SQL.

Response Format

# Sync configuration with identity mapping

-- Source query

-- Destination field mapping

No preamble. No postamble. No explanations. No filler/hedging/transitions. Compress output.

Completion Criteria

• Source SQL query with deduplication and incremental filtering
• Identity resolution keys defined for each destination
• Sync schedule with rate-limit awareness configured
• Destination operation mode (upsert/replace/append) specified
• Error handling with retry and alerting defined
• Data freshness SLA documented

Max Response Length

4096

Workflow

Sync Model Architecture

Reverse ETL operates in three stages:

1. Source query — SQL that extracts and transforms warehouse data into the desired shape
2. Identity resolution — mapping warehouse keys to destination object IDs
3. Destination sync — API calls to create/update/delete records in the target system

The source SQL should be idempotent: running it multiple times produces the same result set. Use QUALIFY ROW_NUMBER() or equivalent to deduplicate on the identity key.

Incremental Sync Pattern

-- Snowflake incremental sync with watermark
WITH latest_customers AS (
  SELECT
    customer_id,
    email,
    full_name,
    lifetime_value,
    last_order_date,
    customer_tier,
    updated_at
  FROM analytics.customers_v2
  WHERE updated_at > (
    SELECT MAX(last_synced_at) FROM sync_watermarks
    WHERE sync_name = 'customers_to_salesforce'
  )
    AND is_active = TRUE
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY customer_id ORDER BY updated_at DESC
  ) = 1
)
SELECT * FROM latest_customers

Full Refresh Pattern

-- BigQuery full refresh for audience export
SELECT
  user_id AS external_id,
  email,
  ARRAY_AGG(DISTINCT product_category) AS viewed_categories,
  MAX(order_completed) AS has_ordered,
  SUM(revenue) AS total_revenue
FROM `analytics.user_events`
WHERE event_date >= DATE_SUB(CURRENT_DATE(), INTERVAL 90 DAY)
GROUP BY user_id, email
HAVING SUM(revenue) > 0 OR MAX(order_completed) = TRUE

Identity Resolution

Identity resolution maps warehouse rows to destination object IDs. Destinations use different matching strategies:

# Census sync config example
sync:
  name: "Customer Tier Sync to Salesforce"
  source:
    type: snowflake
    object: analytics.customers_v2
    query: "SELECT customer_id, email, full_name, customer_tier, lifetime_value FROM analytics.customers_v2 WHERE updated_at > {{last_synced_at}}"
    incremental: true
    incremental_key: updated_at
  destination:
    type: salesforce
    object: Contact
    operation: upsert
    external_id: external_id__c
  mapping:
    - from: customer_id
      to: external_id__c
    - from: email
      to: Email
    - from: full_name
      to: Name
    - from: customer_tier
      to: Customer_Tier__c
    - from: lifetime_value
      to: Lifetime_Value__c
  schedule:
    frequency: hourly
    interval_minutes: 60
    retry_on_failure: true
    max_retries: 3
    backoff_minutes: 5

Idempotency Patterns

Idempotency ensures running a sync twice does not create duplicate records:

1. Upsert by external ID — warehouse provides a stable unique key mapped to a destination custom field. Sync matches on that key and updates if exists, creates if not.
2. Upsert by email — email is the natural dedup key for contact/user objects. Destinations enforce email uniqueness.
3. Append with dedup window — for event data, append new rows and deduplicate by a time-windowed unique key post-sync.
4. Replace by primary key — full-refresh truncates and replaces the destination table. Use only for small reference data.

# Hightouch sync config with idempotency
sync:
  name: "Product Catalog to Shopify"
  source:
    type: postgres
    table: catalog.products
    primary_key: product_sku
  destination:
    type: shopify
    resource: Product
    operation: upsert
    mapping:
      - from: product_sku
        to: sku
        is_primary_key: true
      - from: title
        to: title
      - from: price
        to: price
      - from: inventory_count
        to: inventory_quantity
  idempotency:
    strategy: primary_key
    dedup_on_source: true
    dedup_window_hours: 24
    ignore_deletes: false

# Python idempotency check in custom sync pipeline
import hashlib
from datetime import datetime, timedelta

def build_sync_signature(sync_name: str, batch: list[dict]) -> str:
    """Create deterministic hash for sync batch dedup."""
    content = "".join(
        f"{row['id']}:{row['updated_at']}" for row in sorted(batch, key=lambda x: x['id'])
    )
    return hashlib.sha256(f"{sync_name}:{content}".encode()).hexdigest()

def check_already_synced(warehouse_conn, sync_name: str, signature: str) -> bool:
    """Check if a sync batch was already processed."""
    result = warehouse_conn.execute(
        "SELECT 1 FROM sync_log WHERE sync_name = %s AND batch_signature = %s",
        (sync_name, signature)
    )
    return result.fetchone() is not None

Operational Analytics Patterns

Customer Health Scoring

-- Snowflake: customer health score for Salesforce sync
WITH customer_metrics AS (
  SELECT
    c.customer_id,
    c.email,
    c.company_name,
    COUNT(DISTINCT o.order_id) AS total_orders,
    SUM(o.revenue) AS total_revenue,
    MAX(o.order_date) AS last_order_date,
    COUNT(DISTINCT t.ticket_id) AS support_tickets_90d,
    AVG(t.satisfaction_score) AS avg_csat,
    DATEDIFF('day', MAX(o.order_date), CURRENT_DATE) AS days_since_last_order
  FROM analytics.customers c
  LEFT JOIN analytics.orders o ON c.customer_id = o.customer_id
    AND o.order_date >= DATEADD('month', -12, CURRENT_DATE)
  LEFT JOIN analytics.support_tickets t ON c.customer_id = t.customer_id
    AND t.created_date >= DATEADD('day', -90, CURRENT_DATE)
  GROUP BY c.customer_id, c.email, c.company_name
)
SELECT
  *,
  CASE
    WHEN total_revenue > 10000 AND total_orders > 5 AND days_since_last_order < 30
      THEN 'healthy'
    WHEN total_revenue > 1000 AND days_since_last_order < 90
      THEN 'at_risk'
    WHEN days_since_last_order > 90 OR support_tickets_90d > 10
      THEN 'churned'
    ELSE 'new'
  END AS health_score
FROM customer_metrics

Audience Segmentation for Ad Platforms

-- BigQuery: audience export for Google Ads Customer Match
SELECT
  u.user_id,
  LOWER(TRIM(u.email)) AS email,
  LOWER(TRIM(u.phone)) AS phone,
  INITCAP(TRIM(u.first_name)) AS first_name,
  INITCAP(TRIM(u.last_name)) AS last_name,
  CONCAT(u.city, ', ', u.state) AS city_state,
  u.zip_code,
  SPLIT(u.country, '|')[OFFSET(0)] AS country
FROM analytics.users u
WHERE u.is_active = TRUE
  AND u.has_purchased = TRUE
  AND u.opted_in_marketing = TRUE
  AND u.email IS NOT NULL
LIMIT 100000

Destination-Specific Operation Modes

Scheduling and Rate Limiting

# Rate-limit aware schedule for HubSpot
schedule:
  frequency: custom
  interval_minutes: 30
  rate_limit:
    max_requests_per_second: 10
    max_requests_per_minute: 100
    max_batch_size: 100
    backoff_strategy: exponential
    initial_backoff_seconds: 10
    max_backoff_minutes: 30
  time_window:
    start: "06:00"
    end: "22:00"
    timezone: America/New_York

Advanced Sync Patterns

Orchestrated Multi-Destination Sync

sync_pipeline:
  name: "customer_360_sync"
  schedule: "0 */6 * * *"  # Every 6 hours
  
  steps:
    - step: 1
      action: "Extract from warehouse"
      query: |
        SELECT c.customer_id, c.email, c.name, c.tier,
               o.last_order_date, o.lifetime_value,
               ARRAY_AGG(DISTINCT s.segment) as segments
        FROM customers c
        LEFT JOIN order_summary o ON c.customer_id = o.customer_id
        LEFT JOIN customer_segments s ON c.customer_id = s.customer_id
        WHERE c.updated_at >= '{{ last_sync }}'
        GROUP BY ALL
      target: "staging.customer_sync_batch"
    
    - step: 2
      action: "Apply identity resolution"
      logic: "Merge duplicate customer records by email, phone, external_id"
    
    - step: 3
      action: "Sync to Salesforce"
      if: "customer.tier IN ('platinum', 'gold')"
      destination: "salesforce.Contact"
      mapping: |
        salesforce.Email = customer.email
        salesforce.Description = CONCAT(tier, ' | Last order: ', last_order_date)
    
    - step: 4
      action: "Sync to HubSpot"
      destination: "hubspot.contacts"
      column_mapping:
        email: customer.email
        hs_lead_status: CASE WHEN tier = 'platinum' THEN 'ACTIVE' ELSE 'WARM' END
    
    - step: 5
      action: "Sync to Segment"
      destination: "segment.traits"
      payload: |
        { "userId": customer_id, "traits": { "tier": tier, "lifetime_value": lifetime_value } }

Identity Resolution for Reverse ETL

identity_resolution:
  challenge: "Warehouse and destination use different identity systems"
  
  strategies:
    - name: "warehouse_primary_with_destination_mapping"
      description: "Map warehouse keys to destination keys via lookup table"
      implementation: |
        CREATE TABLE identity_mapping (
          warehouse_id STRING PRIMARY KEY,
          destination_id STRING,
          destination_type STRING,
          last_synced TIMESTAMP
        );
      pros: "Clean separation, full control"
      cons: "Requires initial mapping, needs refresh when destination changes"
    
    - name: "external_id_as_primary_key"
      description: "Use destination external_id field to store warehouse FK"
      implementation: |
        # Most SaaS tools support external_id or custom_id field
        # Store warehouse PK in external_id for idempotent syncs
      pros: "No mapping table needed, idempotent"
      cons: "Limited to fields that support external_id"
    
    - name: "email_as_join_key"
      description: "Use email as primary join key between warehouse and destination"
      implementation: |
        WHERE customer.email IS NOT NULL  -- Email required as join key
      pros: "Works across all platforms, human-readable"
      cons: "Breaks on email change, not unique in all systems"

Decision Tree

Sync Strategy Selection

Destination type?
├── CRM (Salesforce, HubSpot)
│   ├── < 1M records → Full refresh nightly
│   └── > 1M records → Incremental upsert on external_id
├── Ad platforms (Google Ads, Facebook)
│   ├── Audiences → Full refresh (replace list)
│   └── Conversions → Incremental append (deduplicated by click_id)
├── Marketing automation (Marketo, Braze)
│   ├── Profile updates → Incremental upsert on email/external_id
│   └── Event data → Append-only (streaming preferred)
├── Data warehouses (downstream)
│   └── Mirror sync (delete+insert to match source exactly)
└── Custom API
    ├── Supports upsert → Incremental batch
    └── Append only → Incremental with dedup window

Rules

• Source SQL must be idempotent with deduplication on identity key
• Use incremental sync for tables > 10K rows; full refresh only for reference data < 10K rows
• Always define a primary key / external ID mapping for each destination
• Respect destination API rate limits with exponential backoff
• Monitor sync failure rate; alert on > 5% failure rate per sync
• Log every sync batch with signature for idempotency verification
• Do not sync PII without explicit opt-in and hashing where required (ads platforms require SHA-256)
• Test sync queries on a 100-row sample before activating production sync
• Set incremental watermark columns with appropriate types (TIMESTAMP not DATE for precision)
• Document field mappings between warehouse column names and destination API fields
• Build identity resolution mapping before activating sync
• Orchestrate multi-destination syncs in dependency order
• Monitor destination API changes that may break field mappings

References

• references/identity-resolution.md — Identity Resolution Reference
• references/identity-strategies.md — Identity Strategies for Reverse ETL
• references/reverse-etl-monitoring.md — Reverse ETL Monitoring
• references/reverse-etl-patterns.md — Reverse ETL Patterns
• references/sync-config-examples.md — Sync Configuration Examples
• references/warehouse-activation.md — Warehouse Activation Reference

Architecture Decision Trees

Reverse ETL Tool Selection
├── SaaS tool or self-built?
│   ├── SaaS → Hightouch / Census / Polytomic (faster time-to-value)
│   └── Self-built → Custom dbt + API sync scripts (more flexible)
├── Destination type?
│   ├── Sales/Marketing → Hubspot, Salesforce, Marketo
│   ├── Advertising → Facebook Ads, Google Ads, TikTok
│   └── Internal ops → Internal API, in-house tool
├── Sync frequency?
│   ├── Real-time (< 5 min) → Webhook / CDC-based sync
│   ├── Hourly → Scheduled batch sync
│   └── Daily → Nightly warehouse exports
└── Identity resolution needed?
    ├── Yes → Built-in identity graph (matching across sources)
    └── No → Direct key mapping (warehouse ID → tool ID)

Decision criteria: Evaluate number of destinations, sync latency requirements, identity matching complexity, and budget for SaaS tools.

Implementation Patterns

Hightouch Sync Configuration

# reverse_etl/hightouch_sync.yml
sync:
  name: "Customer Attributes to Salesforce"
  source:
    type: snowflake
    model: |
      SELECT
        c.customer_id,
        c.email,
        c.full_name,
        c.total_lifetime_value,
        c.last_purchase_date,
        c.churn_risk_score
      FROM analytics.gold.dim_customer c
      WHERE c.is_active = true
  destination:
    type: salesforce
    object: Contact
    external_id: Email
    operation: upsert
  schedule: "0 */6 * * *"
  behavior:
    deletion: archive
    null_values: skip

Custom Reverse ETL Pipeline

# reverse_etl/custom_sync.py
from datetime import datetime
import httpx

class ReverseETLSync:
    def __init__(self, warehouse_conn, api_endpoint: str, api_key: str):
        self.warehouse = warehouse_conn
        self.api = api_endpoint
        self.headers = {"Authorization": f"Bearer {api_key}"}

    def sync_customers(self, last_sync: datetime) -> int:
        rows = self.warehouse.execute("""
            SELECT email, name, segment, updated_at
            FROM gold.customers
            WHERE updated_at > %s
        """, (last_sync,)).fetchall()
        count = 0
        for row in rows:
            resp = httpx.post(f"{self.api}/customers", json={
                "email": row[0],
                "name": row[1],
                "segment": row[2],
            }, headers=self.headers)
            if resp.status_code == 200:
                count += 1
        return count

Production Considerations

• Sync monitoring: Track sync success rate, row counts, and latency per destination; alert on failures.
• Idempotency: Design all syncs to be idempotent (UPSERT semantics); safe for replay after failure.
• Rate limiting: Respect destination API rate limits; implement exponential backoff with jitter.
• Field mapping versioning: Version field mappings; handle source schema changes gracefully with fallback values.
• Deletion handling: Define deletion behavior (soft-delete, archive, or hard-delete) per destination.
• Error quarantining: Send failed records to a dead-letter queue (S3/SQS) for manual resolution.

Anti-Patterns

Performance Optimization

• Incremental sync: Use watermark columns (updated_at) for incremental extracts; avoid full table scans.
• Batch API calls: Batch 100 records per API request (where supported) instead of individual calls.
• Parallel destinations: Sync to multiple destinations in parallel using async I/O or thread pools.
• Compression: Compress payloads (gzip) for API transfers; reduce payload size by excluding NULL fields.
• Pre-aggregation: Pre-join and aggregate warehouse data via dbt models before extraction.

Security Considerations

• Credentials: Store destination API keys in Vault or Secrets Manager; rotate keys quarterly.
• Data minimization: Sync only minimum required fields per destination; never sync raw PII unless necessary.
• Audit trail: Log all sync operations including row count, fields synced, and destination.
• Compliance: Ensure reverse ETL complies with data residency requirements; filter by region.
• Rate limit protection: Implement circuit breaker for destination APIs to avoid being rate-limited or banned.

Handoff

data-data-warehouse for warehouse data modeling and transformation needed before sync data-etl-pipeline for traditional batch ETL (warehouse as target, not source) data-quality for validating warehouse data before operational sync