用 AWS DMS 将 MySQL 全量 + 增量数据实时同步到 Databend Cloud

CREATE USER 'dms_user'@'%' IDENTIFIED BY 'your_secure_password';
GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'dms_user'@'%';
-- Aurora 还需要额外授权：
GRANT EXECUTE ON PROCEDURE mysql.rds_set_configuration TO 'dms_user'@'%';
FLUSH PRIVILEGES;

-- 设置 binlog 保留 24 小时（根据需要调整）
CALL mysql.rds_set_configuration('binlog retention hours', 24);

-- 方式 A：使用 Access Key / Secret Key
CREATE STAGE IF NOT EXISTS dms_s3_stage
    URL = 's3://dms-databend-bucket/databend/'
    CONNECTION = (
        AWS_KEY_ID     = '<your-access-key>',
        AWS_SECRET_KEY = '<your-secret-key>',
        REGION         = 'ap-southeast-1'
    );

-- 方式 B：使用 IAM Role ARN（生产环境推荐）
CREATE STAGE IF NOT EXISTS dms_s3_stage
    URL = 's3://dms-databend-bucket/databend/'
    CONNECTION = (
        ROLE_ARN = 'arn:aws:iam::123456789012:role/databend-s3-access',
        REGION   = 'ap-southeast-1'
    );

LIST @dms_s3_stage;

CREATE DATABASE IF NOT EXISTS raw;

CREATE TABLE IF NOT EXISTS raw.orders (
    -- DMS 元数据列
    Op                        VARCHAR     NULL,
    _dms_ingestion_timestamp  TIMESTAMP   NULL,

    -- 业务列（与源表 schema 一致）
    order_id                  BIGINT      NULL,
    user_id                   BIGINT      NULL,
    product_id                BIGINT      NULL,
    quantity                  INT         NULL,
    amount                    DECIMAL(12,2) NULL,
    status                    VARCHAR     NULL,
    created_at                TIMESTAMP   NULL,
    updated_at                TIMESTAMP   NULL
) ENABLE_SCHEMA_EVOLUTION = true;

CREATE DATABASE IF NOT EXISTS ods;

CREATE TABLE IF NOT EXISTS ods.orders (
    order_id    BIGINT        NOT NULL,
    user_id     BIGINT        NULL,
    product_id  BIGINT        NULL,
    quantity    INT           NULL,
    amount      DECIMAL(12,2) NULL,
    status      VARCHAR       NULL,
    is_deleted  BOOLEAN       NOT NULL DEFAULT false,   -- 软删除标记
    created_at  TIMESTAMP     NULL,
    updated_at  TIMESTAMP     NULL
);

COPY INTO raw.orders
FROM @dms_s3_stage/orders/
PATTERN = '.*\.parquet'
FILE_FORMAT = (TYPE = PARQUET)
PURGE = TRUE;

INSERT INTO ods.orders
SELECT
    order_id, user_id, product_id, quantity, amount, status,
    false AS is_deleted,
    created_at, updated_at
FROM raw.orders
WHERE Op = 'I';

CREATE STREAM IF NOT EXISTS raw.stream_orders
    ON TABLE raw.orders
    APPEND_ONLY = TRUE;

-- 单表任务示例
CREATE TASK load_orders_raw
    WAREHOUSE = 'default'
    SCHEDULE = 1 MINUTE
    SUSPEND_TASK_AFTER_NUM_FAILURES = 3
AS
    COPY INTO raw.orders
    FROM @dms_s3_stage/orders/
    PATTERN = '.*\.parquet'
    FILE_FORMAT = (TYPE = PARQUET)
    PURGE = TRUE
    ON_ERROR = CONTINUE;

CREATE TASK load_databend_domain_raw
    WAREHOUSE = 'default'
    SCHEDULE = 1 MINUTE
    SUSPEND_TASK_AFTER_NUM_FAILURES = 3
AS
BEGIN
    COPY INTO raw.orders
    FROM @dms_s3_stage/orders/
    PATTERN = '.*\.parquet'
    FILE_FORMAT = (TYPE = PARQUET)
    PURGE = TRUE
    ON_ERROR = CONTINUE;

    -- 同域的其他表在此追加 ...
END;

CREATE TASK merge_orders
      WAREHOUSE = 'default'
      AFTER 'load_orders_raw'
      SUSPEND_TASK_AFTER_NUM_FAILURES = 3
  AS
      MERGE INTO ods.orders AS target
      USING (
          SELECT * FROM (
              SELECT
                  order_id,
                  user_id,
                  product_id,
                  quantity,
                  amount,
                  status,
                  created_at,
                  updated_at,
                  Op                       AS _op,
                  _dms_ingestion_timestamp AS _ts
              FROM raw.stream_orders
          )
          QUALIFY ROW_NUMBER() OVER (
              PARTITION BY order_id
              ORDER BY _ts DESC
          ) = 1
      ) AS source
      ON target.order_id = source.order_id

      WHEN MATCHED AND source._op = 'D' THEN
          UPDATE SET target.is_deleted = true, target.updated_at = source.updated_at

      WHEN MATCHED AND source._op IN ('U', 'I') THEN
          UPDATE SET
              target.user_id    = source.user_id,
              target.product_id = source.product_id,
              target.quantity   = source.quantity,
              target.amount     = source.amount,
              target.status     = source.status,
              target.is_deleted = false,
              target.created_at = source.created_at,
              target.updated_at = source.updated_at

      WHEN NOT MATCHED AND source._op != 'D' THEN
          INSERT (order_id, user_id, product_id, quantity, amount, status, is_deleted, created_at, updated_at)
          VALUES (source.order_id, source.user_id, source.product_id, source.quantity,
                  source.amount, source.status, false, source.created_at, source.updated_at);

-- 第 1 步：启动子任务（MERGE）
ALTER TASK merge_orders RESUME;

-- 第 2 步：启动父任务（COPY INTO）
ALTER TASK load_orders_raw RESUME;

-- 查看任务状态
SHOW TASKS;

-- 查看任务执行历史
SELECT *
FROM TABLE(TASK_HISTORY())
ORDER BY scheduled_time DESC
LIMIT 20;

-- 验证 raw 层数据
SELECT COUNT(*) FROM raw.orders;
SELECT * FROM raw.orders ORDER BY _dms_ingestion_timestamp DESC LIMIT 10;

-- 验证 ods 层数据
SELECT COUNT(*) FROM ods.orders WHERE is_deleted = false;
SELECT * FROM ods.orders ORDER BY updated_at DESC LIMIT 10;

-- 检查 Stream 状态（成功合并后应为空）
SELECT COUNT(*) FROM raw.stream_orders;

INSERT INTO db.orders (user_id, product_id, quantity, amount, status, created_at, updated_at)
VALUES (99999, 88888, 3, 4.4, 'pending', NOW(), NOW());

-- 记录自动生成的 order_id，后续步骤会用到
SELECT LAST_INSERT_ID();

SELECT * FROM ods.orders WHERE order_id = 100000004;
-- 预期：status = 'pending', amount = 4.4, is_deleted = false

UPDATE db.orders
SET status = 'paid', amount = 2.2, updated_at = NOW()
WHERE order_id = 100000004;

SELECT * FROM ods.orders WHERE order_id = 100000004;
-- 预期：status = 'paid', amount = 2.2, is_deleted = false

DELETE FROM db.orders WHERE order_id = 100000004;

SELECT * FROM ods.orders WHERE order_id = 100000004;
-- 预期：is_deleted = true

DELETE FROM ods.orders WHERE order_id = 100000004;
DELETE FROM raw.orders WHERE order_id = 100000004;

-- 暂停任务（先停父任务，再停子任务）
ALTER TASK load_orders_raw SUSPEND;
ALTER TASK merge_orders SUSPEND;

-- 恢复任务（先启子任务，再启父任务）
ALTER TASK merge_orders RESUME;
ALTER TASK load_orders_raw RESUME;

-- 手动触发任务（调试用）
EXECUTE TASK load_orders_raw;

-- 查看未消费的 Stream 数据
SELECT * FROM raw.stream_orders LIMIT 10;

-- 手动清理
DELETE FROM raw.orders
WHERE _dms_ingestion_timestamp < DATEADD(DAY, -15, NOW());

-- 或创建定时清理任务
CREATE TASK cleanup_raw_orders
    WAREHOUSE = 'default'
    SCHEDULE = USING CRON '0 3 * * *'    -- 每天凌晨 3 点执行
AS
    DELETE FROM raw.orders
    WHERE _dms_ingestion_timestamp < DATEADD(DAY, -15, NOW());

ALTER TASK cleanup_raw_orders RESUME;