自动将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"