反向 ETL 与数据激活：让数据仓库的价值回到业务系统

-- models/high_value_customers.sql
SELECT
    c.customer_id,
    c.email,
    c.name,
    c.company,
    SUM(o.revenue)  AS lifetime_value,
    COUNT(o.order_id) AS order_count,
    NTILE(10) OVER (ORDER BY SUM(o.revenue) DESC) AS value_decile
FROM dim.customers c
LEFT JOIN dws.order_fact o ON c.customer_id = o.customer_id
GROUP BY 1, 2, 3, 4
HAVING SUM(o.revenue) >= 50000

# census_sync_salesforce.yml
sync:
  name: high_value_customers_to_crm
  source:
    model: high_value_customers
    warehouse: snowflake
  destination:
    type: salesforce
    object: Account
    operation: upsert
    external_id: account_customer_id__c
  mapping:
    - {from: customer_id, to: account_customer_id__c}
    - {from: name,        to: name}
    - {from: lifetime_value, to: lifetime_value__c}
    - {from: value_decile, to: value_decile__c}
  schedule: daily_at_0200

# intercom_sync.py
import requests

def sync_customer_360(rows):
    for r in rows:
        resp = requests.post(
            "https://api.intercom.io/contacts",
            headers={"Authorization": f"Bearer {INTERCOM_TOKEN}"},
            json={
                "email": r["email"],
                "custom_attributes": {
                    "lifetime_value": r["lifetime_value"],
                    "tier": r["tier"],
                    "churn_risk": r["churn_risk_score"],
                },
                "tags": [{"name": r["segment"]}],
            },
        )
        resp.raise_for_status()

-- models/segment_churn_risk.sql
-- 流失风险分群：近 90 天无购买且登录下降
SELECT
    u.user_id,
    u.email,
    'high_risk' AS segment,
    u.churn_score
FROM dws.user_health u
WHERE u.days_since_last_purchase > 90
  AND u.login_trend_30d < 0.3
  AND u.is_active = true;

{
  "sync_run": "sync_20260927_0200",
  "model": "segment_churn_risk",
  "destination": "braze",
  "members_added": 12340,
  "members_removed": 210,
  "members_unchanged": 45520,
  "duration_sec": 86,
  "status": "success"
}

POST /webhooks/custom-activation HTTP/1.1
Content-Type: application/json

{
  "event": "user_activated",
  "user": {
    "id": "u_10086",
    "email": "user@example.com",
    "segment": "high_value",
    "score": 0.94,
    "ts": "2026-09-27T03:00:00Z"
  }
}

#!/bin/bash
# trigger_sync.sh
curl -X POST "https://api.hightouch.com/api/v1/syncs/run" \
  -H "Authorization: Bearer $HIGHTOUCH_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
        "syncId": "sync_high_value",
        "runImmediately": true
      }'

-- 度量：同步后的转化提升
SELECT
    act.segment,
    COUNT(DISTINCT act.user_id) AS activated_users,
    COUNT(DISTINCT conv.order_id) AS converted_orders,
    SUM(conv.revenue) AS revenue,
    ROUND(COUNT(DISTINCT conv.order_id) * 1.0 / NULLIF(COUNT(DISTINCT act.user_id), 0), 3) AS cvr
FROM dwh.reverse_etl_activation act
LEFT JOIN dwh.orders conv ON act.user_id = conv.user_id
  AND conv.created_at BETWEEN act.activated_at AND act.activated_at + INTERVAL '14 days'
GROUP BY 1;

-- models/identity_map.sql
SELECT DISTINCT
    c.customer_id,
    COALESCE(crm.account_id, '') AS crm_id,
    c.email,
    c.phone
FROM dim.customers c
LEFT JOIN staging.crm_account_map crm ON c.customer_id = crm.customer_id;

# realtime_activation.yaml
stream:
  source: kafka
  topic: cdc.dwh.user_health
  transform:
    - filter: "payload.op in ('c','u')"
    - enrich: "join dim.customers on user_id"
  sink:
    type: rudderstack
    destination: braze
    event: UserHealthUpdated
    id: user_id

# hybrid_activation.py
def route_activation(user: dict):
    if user["tier"] == "platinum" or user["churn_score"] > 0.9:
        # 高优先级走实时通道
        push_realtime(user)
    else:
        # 其余进入批量队列，次日同步
        enqueue_batch(user)

# governance_policy.yaml
governance:
  field_masking:
    - field: phone
      mask: "keep_last_4"
    - field: email
      mask: "hash_sha256"
  audit:
    log_sync_runs: true
    log_schema_changes: true
  rollout:
    canary_percent: 5
    auto_rollback_on_error_rate_gt: 0.01
  approval:
    required_for: [payment, compliance]

# rollback.py
def rollback_sync(sync_id: str, model_snapshot: str):
    # 1. 暂停该同步的定时触发
    pause_sync(sync_id)
    # 2. 用上一个可信快照重建目标数据
    rebuild_destination(model_snapshot)
    # 3. 审计并通知受影响业务方
    notify_stakeholders(sync_id, action="rollback")