全部产品
Search

搜索本产品

    云原生数据库 PolarDB：AutoETL 参数配置和实践案例

    文档中心

    云原生数据库 PolarDB：AutoETL 参数配置和实践案例

    更新时间：Aug 27, 2026

    AutoETL提供多个配置选项，支持您按业务场景调整同步行为，例如JSON字段转换、搜索路由、写入模式以及链路使用的计算资源等。本文介绍搜索视图和ETL存储过程两种配置入口下的参数写法，并提供常见场景的最佳实践案例。

    参数在两种入口下的写法如下：

    • 搜索视图：以WITH (...)子句内联在DDL中，仅对当前搜索视图生效。

    • ETL存储过程：兼容Flink语法。其中链路配置可通过session变量esl_link_options设置，同步参数（源表读取、目标写入）则直接写在Flink SQL的WITH (...)子句中。

    计算资源配置

    ETL链路通过一个或多个Worker工作线程来完成实际的传输工作。您可以通过下列参数控制链路使用的资源：

    参数

    说明

    默认值

    parallelism

    当前ETL链路使用的总并发数。

    4

    link.tm.cpu

    每个Worker分配的CPU核数。

    2

    link.tm.slot

    每个Worker支持的任务并发数。

    4

    说明

    Worker分配支持的并发数不是越高越好，需要考虑Worker CPU总量。

    传输链路以CU计量，CU计算公式为CU = parallelism / link.tm.slot * link.tm.cpu + 1。下述案例配置对应CU数为5。

    搜索视图

    CREATE SEARCH VIEW view_test
WITH (
  'parallelism' = '8',
  'link.tm.cpu' = '4',
  'link.tm.slot' = '8'
) AS SELECT * FROM t1;

    ETL存储过程

    SET esl_link_options = "'parallelism' = '8', 'link.tm.cpu' = '4', 'link.tm.slot' = '8'";
CALL dbms_etl.sync_by_sql("search", "

CREATE TEMPORARY TABLE `db1`.`t1` (
  `id`   BIGINT,
  `c1`   STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'database-name' = 'db1',
  'table-name' = 't1'
);

CREATE TEMPORARY TABLE `dest` (
  `id`  BIGINT,
  `c1`  STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'opensearch',
  'index' = 'dest'
);

INSERT INTO `dest`
SELECT * FROM `db1`.`t1`;
");

    支持的参数列表

    下表汇总AutoETL目前支持的全部参数。

    链路参数

    参数

    说明

    默认值

    parallelism

    当前ETL链路使用的总并发数。

    4

    link.tm.cpu

    每个Worker分配的CPU核数。

    2

    link.tm.slot

    每个Worker支持的任务并发数。

    4

    link.param.execution.checkpointing.interval

    ETL链路的checkpoint周期。

    180s

    同步参数

    参数

    说明

    默认值

    gdn-cluster

    同步的目标GDN集群。

    当前实例默认集群

    sink-name

    搜索视图在PolarSearch中的索引名称（存储过程通过目标表'index'指定）。

    搜索视图名

    scan.incremental.snapshot.chunk.size

    全量扫描阶段每个chunk的行数，影响全量切片粒度与并发效率。

    131072

    routing-fields

    用于将文档路由到PolarSearch指定分片的字段名，多字段用;分隔。

    无

    ignore-delete

    是否忽略删除操作，true时链路不再向目标索引下发delete。

    false

    sink.force-index-request

    是否强制以Index模式整行替换写入（否则为Update部分字段更新）。

    false

    sink.json-flatten.fields

    需要解析JSON的字段名，多字段用;分隔。

    无

    sink.json-flatten.mode

    JSON解析模式：nested（默认，转nested对象）/ flatten（按第一层展平）。

    nested

    最佳实践案例

    JSON字段自动转换

    当前MySQL的JSON字段同步到Search中默认以字符串形式存储。PolarDB搜索视图支持自动将MySQL的JSON字段在同步到PolarSearch时解析，生成nested字段，或者对JSON字段进行展平。

    1. 数据准备

      在集群中执行以下SQL，创建示例数据库和表并插入测试数据：

      CREATE DATABASE IF NOT EXISTS db1;
USE db1;
CREATE TABLE IF NOT EXISTS t1 (
    id INT PRIMARY KEY,
    c1 JSON
);
INSERT INTO t1(id, c1) VALUES
(1, '{"age": 75, "name": "User_A5pqo", "tags": ["q7XG", "Unx9", "EBy8"], "active": false, "metadata": {"source": "script", "version": "1.0", "created_at": "2026-03-06T07:04:45.264573Z"}}'),
(2, '{"age": 55, "name": "User_xL1YH", "tags": ["QNcC", "kqU7"], "active": true, "metadata": {"source": "script", "version": "1.0", "created_at": "2026-03-06T07:04:45.264632Z"}}'),
(3, '{"age": 25, "name": "User_zoRSH", "tags": [ ], "active": true, "metadata": {"source": "script", "version": "1.0", "created_at": "2026-03-06T07:04:45.264654Z"}}');

    2. 默认配置（字符串存储）

      搜索视图

      CREATE SEARCH VIEW json_test AS SELECT * FROM t1;

      ETL存储过程

      CALL dbms_etl.sync_by_sql("search", "

CREATE TEMPORARY TABLE `db1`.`t1` (
  `id`   INT,
  `c1`   STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'database-name' = 'db1',
  'table-name' = 't1'
);

CREATE TEMPORARY TABLE `dest` (
  `id`  INT,
  `c1`  STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'opensearch',
  'index' = 'json_test'
);

INSERT INTO `dest`
SELECT * FROM `db1`.`t1`;
");

      验证数据

      {
  "_index" : "json_test",
  "_id" : "3",
  "_source" : {
    "id" : 3,
    "c1" : "{\"age\":25,\"name\":\"User_zoRSH\",\"tags\":[ ],\"active\":true,\"metadata\":{...}}"
  }
}

    3. 将c1转为nested类型

      搜索视图

      CREATE SEARCH VIEW json_test
WITH (
  'sink.json-flatten.fields' = 'c1'   -- 多个字段用 ; 分隔
) AS SELECT * FROM t1;

      ETL存储过程

      CALL dbms_etl.sync_by_sql("search", "

CREATE TEMPORARY TABLE `db1`.`t1` (
  `id`   INT,
  `c1`   STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'database-name' = 'db1',
  'table-name' = 't1'
);

CREATE TEMPORARY TABLE `dest` (
  `id`  INT,
  `c1`  STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'opensearch',
  'index' = 'json_test',
  'sink.json-flatten.fields' = 'c1'   -- 多个字段用 ; 分隔
);

INSERT INTO `dest`
SELECT * FROM `db1`.`t1`;
");

      验证数据

      {
  "_index" : "json_test",
  "_id" : "3",
  "_source" : {
    "id" : 3,
    "c1" : {
      "age" : 25,
      "name" : "User_zoRSH",
      "tags" : [ ],
      "active" : true,
      "metadata" : { "source" : "script", "version" : "1.0", "created_at" : "2026-03-06T07:04:45.264654Z" }
    }
  }
}

    4. 将c1按JSON第一层展平

      搜索视图

      CREATE SEARCH VIEW json_test
WITH (
  'sink.json-flatten.fields' = 'c1',
  'sink.json-flatten.mode' = 'flatten'
) AS SELECT * FROM t1;

      ETL存储过程

      CALL dbms_etl.sync_by_sql("search", "

CREATE TEMPORARY TABLE `db1`.`t1` (
  `id`   INT,
  `c1`   STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'database-name' = 'db1',
  'table-name' = 't1'
);

CREATE TEMPORARY TABLE `dest` (
  `id`  INT,
  `c1`  STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'opensearch',
  'index' = 'json_test',
  'sink.json-flatten.fields' = 'c1',
  'sink.json-flatten.mode' = 'flatten'
);

INSERT INTO `dest`
SELECT * FROM `db1`.`t1`;
");

      验证数据：展平后，JSON第一层的字段成为PolarSearch索引中的独立顶级字段。

      {
  "_index" : "json_test",
  "_id" : "3",
  "_source" : {
    "id" : 3,
    "metadata" : { "source" : "script", "version" : "1.0", "created_at" : "2026-03-06T07:04:45.264654Z" },
    "name" : "User_zoRSH",
    "active" : true,
    "age" : 25,
    "tags" : [ ]
  }
}

    设定Search路由字段

    指定一个或多个搜索视图中的字段名称，将某一行的数据路由到PolarSearch的指定分片中。

    1. 数据准备

      CREATE DATABASE IF NOT EXISTS db1;
USE db1;
CREATE TABLE IF NOT EXISTS t1 (
    id INT PRIMARY KEY,
    c1 BIGINT
);
INSERT INTO t1(id, c1) VALUES (1, 3), (2, 2), (3, 1);

    2. 配置路由字段

      搜索视图

      CREATE SEARCH VIEW routing_test
WITH (
  'routing-fields' = 'c1'   -- 多个字段用 ; 分隔
) AS SELECT * FROM t1;

      ETL存储过程

      CALL dbms_etl.sync_by_sql("search", "

CREATE TEMPORARY TABLE `db1`.`t1` (
  `id`   INT,
  `c1`   BIGINT,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'database-name' = 'db1',
  'table-name' = 't1'
);

CREATE TEMPORARY TABLE `dest` (
  `id`  INT,
  `c1`  BIGINT,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'opensearch',
  'index' = 'routing_test',
  'routing-fields' = 'c1'   -- 多个字段用 ; 分隔
);

INSERT INTO `dest`
SELECT * FROM `db1`.`t1`;
");

      验证数据：文档按c1字段的值路由到对应分片，_routing字段记录了路由依据的值。

      { "_index" : "routing_test", "_id" : "1", "_routing" : "3", "_source" : { "id" : 1, "c1" : 3 } },
{ "_index" : "routing_test", "_id" : "3", "_routing" : "1", "_source" : { "id" : 3, "c1" : 1 } },
{ "_index" : "routing_test", "_id" : "2", "_routing" : "2", "_source" : { "id" : 2, "c1" : 2 } }

    忽略删除

    对于多表汇聚的搜索视图，AutoETL服务对目标索引的更新模式为先删除后插入。如果您不希望在查询时访问到被删除数据的中间状态，可以启用忽略删除，让链路在同步时跳过删除动作。

    搜索视图

    CREATE SEARCH VIEW view_test
WITH (
  'ignore-delete' = 'true'
) AS SELECT * FROM t1;

    ETL存储过程

    CALL dbms_etl.sync_by_sql("search", "

CREATE TEMPORARY TABLE `db1`.`t1` (
  `id`   BIGINT,
  `c1`   STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'database-name' = 'db1',
  'table-name' = 't1'
);

CREATE TEMPORARY TABLE `dest` (
  `id`  BIGINT,
  `c1`  STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'opensearch',
  'index' = 'view_test',
  'ignore-delete' = 'true'
);

INSERT INTO `dest`
SELECT * FROM `db1`.`t1`;
");

    此时链路不再执行删除动作。

    说明

    由于不会清理数据，可能会造成PolarSearch索引膨胀。建议在MySQL源表用字段进行删除标记，同步到PolarSearch后，通过定时任务清理标记删除的文档。

    替换写入

    当前AutoETL服务在写入PolarSearch索引时，如果文档已存在，会采用Update模式，只更新搜索视图写入的字段。对于业务需要整行替换写入的场景，可配置sink.force-index-request启用Index写入模式。

    搜索视图

    CREATE SEARCH VIEW view_test
WITH (
  'sink.force-index-request' = 'true'
) AS SELECT * FROM t1;

    ETL存储过程

    CALL dbms_etl.sync_by_sql("search", "

CREATE TEMPORARY TABLE `db1`.`t1` (
  `id`   BIGINT,
  `c1`   STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'mysql',
  'database-name' = 'db1',
  'table-name' = 't1'
);

CREATE TEMPORARY TABLE `dest` (
  `id`  BIGINT,
  `c1`  STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'opensearch',
  'index' = 'view_test',
  'sink.force-index-request' = 'true'
);

INSERT INTO `dest`
SELECT * FROM `db1`.`t1`;
");