data-reverse-etlUse this skill when asked about reverse ETL, Census, Hightouch, Grouparoo, operational analytics, syncing data from warehouse to SaaS, warehosue-to-SaaS, operational data activation, data warehouse as source, sync configs, idempotent syncs, audience export, or data activation. This skill enforces: warehouse-first sync architecture with idempotent writes, in…
Independent GitHub snapshot · Automated classification · Original documentation governs
Sync aggregated, transformed data from a data warehouse to operational SaaS tools (CRM, marketing, support) to power customer-facing and internal operational workflows.
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".
Before activating, verify:
Reverse ETL pipeline config with SQL source query, sync schedule, identity mapping, and destination-specific operation config as YAML and SQL.
# Sync configuration with identity mapping
-- Source query
-- Destination field mapping
No preamble. No postamble. No explanations. No filler/hedging/transitions. Compress output.
4096
Reverse ETL operates in three stages:
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.
-- 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
-- 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 maps warehouse rows to destination object IDs. Destinations use different matching strategies:
| Destination | Match Key | Sync Mode | Dedup Strategy |
|---|---|---|---|
| Salesforce | external_id__c or email | Upsert | Custom External ID field |
| HubSpot | email or hs_object_id | Upsert | Email dedup rule |
| Marketo | email or foreignKey | Upsert | Email dedup |
| Braze | external_id or alias_name | Merge | External ID |
| Amplitude | user_id or device_id | Merge | User ID merge |
| Zendesk | email or external_id | Upsert | External ID |
| Google Ads | hashed_email or hashed_phone | Upload (CUID) | SHA-256 hashing |
| Facebook Audiences | email or pixel_id | Upload (CAPI) | SHA-256 hashing |
# 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 ensures running a sync twice does not create duplicate records:
# 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
-- 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
-- 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
| Mode | Behavior | Use Case |
|---|---|---|
upsert | Create or update by match key | Customer profiles, CRM contacts |
append | Always insert new rows | Event data, audit logs |
replace | Truncate destination and insert all rows | Small reference tables |
merge | Match and merge fields (Braze, Amplitude) | User identity resolution |
mirror | Full reconciliation: insert, update, delete to match source exactly | Product catalogs, inventory |
# 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
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:
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"
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
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.
# 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
# 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
| Anti-Pattern | Consequence | Solution |
|---|---|---|
| Syncing all columns to every destination | PII leakage, high bandwidth | Only sync required fields per destination |
| No dedup before sync | Duplicate records in CRM | Deduplicate on source identity key |
| Full resync every time | API quota exhaustion | Only sync changed records since last sync |
| Ignoring destination schema limits | Truncated/failed records | Validate length and format before sync |
| No identity resolution | Orphaned records in destination | Pre-merge identities before sync |
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