全部產品
Search

搜尋本產品

    PolarDB：AutoETL 參數配置和實踐案例

    文件中心

    PolarDB：AutoETL 參數配置和實踐案例

    更新時間：Aug 28, 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`;
");