Trình Kết Nối Snowflake

connectors:
  snowflake:
    enabled: true
    account: "${SNOWFLAKE_ACCOUNT}"
    username: "${SNOWFLAKE_USER}"
    private_key_path: "${SNOWFLAKE_PRIVATE_KEY_PATH}"
    warehouse: "COMPUTE_WH"
    database: "ANALYTICS"
    schema: "PUBLIC"

    sync_models:
      - name: "brevo_contacts"
        query: |
          SELECT email, first_name, last_name, plan, ltv
          FROM customers
          WHERE is_active = true
        target: brevo_contacts
        primary_key: email
        schedule: "0 2 * * *"  # Hàng ngày lúc 2 giờ sáng

field_mapping:
  email: email
  first_name: FIRSTNAME
  last_name: LASTNAME
  plan_tier: PLAN_TYPE
  lifetime_value: LIFETIME_VALUE
  last_login: LAST_LOGIN_DATE
  is_paid: IS_PAID_CUSTOMER
  product_score: ENGAGEMENT_SCORE

-- Lấy customers hoạt động với tất cả thuộc tính cần thiết
SELECT
  c.email,
  c.first_name,
  c.last_name,
  c.phone,
  s.plan_name as plan,
  s.mrr,
  c.created_at as signup_date,
  DATEDIFF('day', MAX(e.event_time), CURRENT_TIMESTAMP()) as days_since_active
FROM customers c
LEFT JOIN subscriptions s ON c.id = s.customer_id AND s.status = 'active'
LEFT JOIN events e ON c.id = e.user_id AND e.event_time >= DATEADD('day', -30, CURRENT_DATE())
WHERE c.email IS NOT NULL
  AND c.deleted_at IS NULL
GROUP BY 1,2,3,4,5,6,7

import { TajoClient } from '@tajo/sdk';

const tajo = new TajoClient({
  apiKey: process.env.TAJO_API_KEY,
  brevoApiKey: process.env.BREVO_API_KEY
});

await tajo.connectors.connect('snowflake', {
  account: process.env.SNOWFLAKE_ACCOUNT,
  username: process.env.SNOWFLAKE_USER,
  privateKeyPath: process.env.SNOWFLAKE_PRIVATE_KEY_PATH,
  warehouse: process.env.SNOWFLAKE_WAREHOUSE,
  database: process.env.SNOWFLAKE_DATABASE
});

// Đồng bộ kết quả query sang Brevo
await tajo.connectors.reverseETL('snowflake', {
  model: 'brevo_contacts',
  mode: 'upsert',
  primaryKey: 'email'
});