PolarDB：自動將PolarDB MySQL版資料同步至PolarSearch

資料準備

在PolarDB MySQL版中執行以下SQL語句，建立樣本資料庫和表，並插入測試資料。

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, 'apple', 'red'),
(2, 'banana', 'yellow'),
(3, 'grape', 'purple');

建立同步鏈路

使用dbms_etl.sync_by_table預存程序，建立從db1.t1表到PolarSearch節點的索引dest的同步任務。

文法

call dbms_etl.sync_by_table("search", "<source_table>", "<sink_table>", "<column_list>");

參數說明

使用限制

• 源表需包含主鍵或唯一鍵。

• 在不同的同步鏈路中，不能使用相同的源表或目標表。

• 建立鏈路後，源表新增的列預設不會被自動同步。如需同步新增列，請重建鏈路。

• 如果您希望使用自己定義的目標索引配置，您可以先在PolarSearch節點中手動建立索引並定義其配置，然後再建立同步鏈路。如果鏈路建立時目標索引不存在，系統將自動建立。

樣本

• 將db1.t1全表同步到PolarSearch的dest索引：

  call dbms_etl.sync_by_table("search", "db1.t1", "dest", "");

• 將db1.t1表的c1和c2列同步到dest索引：

  call dbms_etl.sync_by_table("search", "db1.t1", "dest", "c1, c2");

驗證資料

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

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

資料準備

在PolarDB MySQL版中執行以下SQL語句，建立樣本資料庫和表，並插入測試資料。

CREATE DATABASE IF NOT EXISTS db1;
CREATE DATABASE IF NOT EXISTS db2;
CREATE DATABASE IF NOT EXISTS db3;

CREATE TABLE IF NOT EXISTS db1.t1 (id INT PRIMARY KEY, c1 INT);
CREATE TABLE IF NOT EXISTS db2.t2 (id INT PRIMARY KEY, c2 INT);
CREATE TABLE IF NOT EXISTS db3.t3 (id INT PRIMARY KEY, c3 VARCHAR(10));

INSERT INTO db1.t1(id, c1) VALUES (1, 11), (2, 22), (3, 33);
INSERT INTO db2.t2(id, c2) VALUES (1, 111), (2, 222), (4, 444);
INSERT INTO db3.t3(id, c3) VALUES (1, 'aaa'), (3, 'ccc'), (4, 'ddd');

建立同步鏈路

使用dbms_etl.sync_by_map預存程序，可將多個表的資料連線（JOIN）後，匯聚到一個PolarSearch節點的索引中。

文法

call dbms_etl.sync_by_map(
    "search",
    "<columns_map>", -- 目標索引欄位與源表欄位的映射關係
    "<join_fields>", -- 表之間的串連鍵
    "<join_types>",  -- 連線類型 (inner, left)
    "<filter>"       -- 資料過濾條件
);

參數說明

使用限制

• 所有參與同步的源表必須包含主鍵。

• 該功能使用流式計算，同步過程中僅保證最終一致性。

• 對於目標索引的更新模式為先刪除後插入。如果您不希望在查詢時訪問到被刪除資料的中間狀態，可以在執行命令前設定會話變數set sink_options = "'ignore-delete' = 'true'";以忽略PolarSearch節點資料刪除的選項。

樣本

• 兩張表INNER JOIN：將db1.t1和db2.t2通過id欄位進行INNER JOIN，並將t1.c1和t2.c2同步到dest索引的c1和c2欄位。

  call dbms_etl.sync_by_map(
    "search",
    "dest.c1(db1.t1.c1), dest.c2(db2.t2.c2)",
    "dest.id=db1.t1.id,db2.t2.id", 
    "inner",
     ""
  );

• 多張表混合JOIN並過濾：db1.t1、db2.t2和db3.t3三張表串連，其中t1與t2為INNER JOIN，t1與t3為LEFT JOIN，並篩選t1.c1 > 10且t2.c2 < 100的資料。

  call dbms_etl.sync_by_map(
    "search", 
    "dest.c1(db1.t1.c1), dest.c2(db2.t2.c2), dest.c3(db3.t3.c3)", 
    "dest.id=db1.t1.id,db2.t2.id,db3.t3.id", 
    "inner,left", 
    "db1.t1.c1 > 10 and db2.t2.c2 < 100"
  );

驗證資料

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

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

建立同步鏈路

對於需要複雜轉換、彙總或計算的情境，dbms_etl.sync_by_sql預存程序支援使用Flink SQL文法定義資料同步邏輯。

文法

call dbms_etl.sync_by_sql("search", "<sync_sql>");

樣本

call dbms_etl.sync_by_sql("search", "
-- 步驟1：定義 PolarDB 源表
CREATE TEMPORARY TABLE `db1`.`sbtest1` (
  `id`   BIGINT,
  `k`    BIGINT,
  `c`    STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'hostname' = 'xxxxxxx', -- 填寫 PolarDB 叢集地址
  'port' = '3306',
  'username' = 'xxx',     -- 生產環境嚴禁使用明文
  'password' = 'xxx',     -- 生產環境嚴禁使用明文
  'database-name' = 'db1',
  'table-name' = 'sbtest1'
);

-- 步驟2：定義 PolarSearch 目標表
CREATE TEMPORARY TABLE `dest` (
  `k`  BIGINT,
  `max_c` STRING,
  PRIMARY KEY (`k`) NOT ENFORCED
) WITH (
  'connector' = 'opensearch',
  'hosts' = 'xxxxxx:xxxx',     -- 填寫 PolarSearch 串連地址
  'index' = 'dest',
  'username' = 'xxx',     -- 生產環境嚴禁使用明文
  'password' = 'xxx'      -- 生產環境嚴禁使用明文
);

-- 步驟3：定義計算和插入邏輯
INSERT INTO `dest`
SELECT
    `t1`.`k`,
    MAX(`t1`.`c`)
FROM `db1`.`sbtest1` AS `t1`
GROUP BY `t1`.`k`;
");

驗證資料

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

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