AutoETL多叢集資料匯聚 - PolarDB

AutoETL支援匯聚多個PolarDB MySQL版叢集的業務資料到同一個PolarSearch節點中，便於您對分布在不同叢集中的資料進行統一檢索和分析。本文介紹如何通過AutoETL建立多叢集資料匯聚鏈路。

說明

當前功能處於灰階階段。如您有相關需求，請提交工單與我們聯絡，以便為您開啟該功能。

背景資訊

在實際業務情境中，資料可能分布在多個PolarDB MySQL版叢集中。當您需要對這些分散的資料進行統一的全文檢索索引或分析時，可以通過AutoETL的多叢集資料匯聚功能，將多個源叢集的資料同步到一個匯聚叢集的PolarSearch節點中。

整個資料流轉過程如下：

在源叢集中，通過CREATE ETL GRANT授權匯聚叢集訪問本叢集的資料。
在匯聚叢集中，通過ETL預存程序（dbms_etl.sync_by_sql）建立匯聚鏈路，在源表的WITH子句中通過polardb-mysql-instance參數指定資料來源的PolarDB MySQL版叢集。
鏈路啟動後，AutoETL鏈路會檢查每個源叢集是否已授權匯聚叢集的ETL訪問。確認授權後，引擎從各源叢集讀取資料並同步到匯聚叢集的PolarSearch節點中。

使用限制

所有參與匯聚的PolarDB MySQL版叢集必須屬於同一個阿里雲帳號（UID）。
目前僅支援通過ETL預存程序（dbms_etl.sync_by_sql）建立多叢集匯聚鏈路，不支援搜尋視圖方式。
刪除ETL授權不影響已建立且正在啟動並執行同步鏈路。

步驟一：建立ETL授權

為了允許匯聚叢集同步源叢集的資料，您需要在每個源叢集中建立ETL授權，授權匯聚叢集通過AutoETL訪問本叢集的資料。

建立ETL授權

在源叢集中執行以下SQL語句，通過<allow_instance_id>指定允許ETL同步資料的匯聚叢集ID。

CREATE ETL GRANT `<allow_instance_id>`;

刪除ETL授權

DROP ETL GRANT `<allow_instance_id>`;

查看ETL授權

SHOW ETL GRANTS;

步驟二：建立匯聚鏈路

您可以通過ETL預存程序（dbms_etl.sync_by_sql）建立多叢集的匯聚鏈路。在源表的WITH子句中，通過polardb-mysql-instance參數指定該表所在的源叢集ID。

準備資料

假設有兩個PolarDB MySQL版叢集（叢集A和叢集B），需要將叢集A的資料匯聚到叢集B的PolarSearch節點中。

在叢集A中建立測試資料，並授權叢集B進行ETL同步：

-- 在叢集A中執行
CREATE DATABASE IF NOT EXISTS db1;
USE db1;
CREATE TABLE IF NOT EXISTS t1 (
    id INT PRIMARY KEY,
    c1 VARCHAR(100),
    c2 VARCHAR(100)
);
INSERT INTO t1(id, c1, c2) VALUES
(1, '1', '1'),
(2, '1', '1'),
(3, '1', '1');

-- 授權叢集B訪問本叢集資料
-- 將pc-xxx替換為叢集B的實際叢集ID
CREATE ETL GRANT `pc-xxx`;

在叢集B中建立測試資料：

-- 在叢集B中執行
CREATE DATABASE IF NOT EXISTS db2;
USE db2;
CREATE TABLE IF NOT EXISTS t2 (
    id INT PRIMARY KEY,
    c1 VARCHAR(100),
    c2 VARCHAR(100)
);
INSERT INTO t2(id, c1, c2) VALUES
(1, '2', '2'),
(2, '2', '2'),
(3, '2', '2');

定義匯聚鏈路

在叢集B中執行以下SQL語句，建立匯聚鏈路將兩個叢集的資料同步到叢集B的PolarSearch節點中。

說明

源表和目標表的串連資訊（如串連地址、連接埠、帳號密碼等）由系統自動設定，您無需在WITH子句中手動指定。僅需通過polardb-mysql-instance參數指定源表所在的叢集ID。

-- 在叢集B中執行
CALL dbms_etl.sync_by_sql("search", "

-- 步驟1：定義叢集A的源表
CREATE TEMPORARY TABLE `db1`.`t1` (
  `id`   BIGINT,
  `c1`   STRING,
  `c2`   STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'database-name' = 'db1',
  'table-name' = 't1',
  'polardb-mysql-instance' = 'pc-xxx'  -- 替換為叢集A的實際叢集ID
);

-- 步驟2：定義叢集B的源表
CREATE TEMPORARY TABLE `db2`.`t2` (
  `id`   BIGINT,
  `c1`   STRING,
  `c2`   STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'database-name' = 'db2',
  'table-name' = 't2',
  'polardb-mysql-instance' = 'pc-xxx'  -- 替換為叢集B的實際叢集ID
);

-- 步驟3：定義PolarSearch目標表
CREATE TEMPORARY TABLE `dest` (
  `id`  BIGINT,
  `p1_c1` STRING,
  `p2_c1` STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'opensearch',
  'index' = 'dest'
);

-- 步驟4：定義計算和插入邏輯
INSERT INTO `dest`
SELECT
    `t1`.`id`,
    `t1`.`c1`,
    `t2`.`c1`
FROM `db1`.`t1` AS `t1`
LEFT JOIN `db2`.`t2` AS `t2`
  ON `t1`.`id` = `t2`.`id`;
");

步驟三：驗證資料

串連到匯聚叢集的PolarSearch節點，使用與Elasticsearch相容的REST API進行查詢，確認資料已同步。

# 將<user>:<password>替換為PolarSearch節點的帳號密碼
# 將<polarsearch_endpoint>替換為PolarSearch節點的串連地址與連接埠
curl -u <user>:<password> -X GET "http://<polarsearch_endpoint>/dest/_search"

PolarDB：AutoETL多叢集資料匯聚