Configuração de parâmetros e melhores práticas do AutoETL - PolarDB: Central de Documentação do Alibaba Cloud

Configure os parâmetros em dois pontos de entrada:

Search view: Defina os parâmetros diretamente na cláusula WITH (...) do DDL. Esses parâmetros afetam apenas a search view atual.
ETL stored procedure: Compatível com a sintaxe do Flink. Defina a configuração do link pela variável de sessão esl_link_options e escreva os parâmetros de sincronização (leitura da tabela de origem e escrita no destino) diretamente na cláusula WITH (...) do Flink SQL.

Configuração de recursos computacionais

Um link de ETL usa uma ou mais threads de trabalho para transferir dados. Controle os recursos do link pelos seguintes parâmetros:

Parâmetro	Descrição	Valor padrão
`parallelism`	Concorrência total do link de ETL atual.	4
`link.tm.cpu`	Núcleos de CPU alocados para cada worker.	2
`link.tm.slot`	Concorrência de tarefas por worker.	4

Nota

Maior concorrência de workers nem sempre é melhor. Considere também a capacidade total de CPU de todos os workers.

A medição do link de transferência ocorre em CU. A fórmula é CU = parallelism / link.tm.slot * link.tm.cpu + 1. O exemplo de configuração a seguir corresponde a 5 CUs.

Search view

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

ETL stored procedure

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`;
");

Lista de parâmetros compatíveis

As tabelas a seguir resumem todos os parâmetros compatíveis com o AutoETL atualmente.

Parâmetros de link

Parâmetro	Descrição	Valor padrão
`parallelism`	Concorrência total do link de ETL atual.	4
`link.tm.cpu`	Núcleos de CPU alocados para cada worker.	2
`link.tm.slot`	Concorrência de tarefas por worker.	4
`link.param.execution.checkpointing.interval`	Intervalo de checkpoint do link de ETL.	180s

Parâmetros de sincronização

Parâmetro	Descrição	Valor padrão
`gdn-cluster`	Cluster GDN de destino da sincronização.	Cluster padrão da instância atual
`sink-name`	Nome do índice da search view no PolarSearch. Em stored procedures, especifique a tabela de destino por `'index'`.	Nome da search view
`scan.incremental.snapshot.chunk.size`	Linhas por chunk durante a fase de varredura completa. Afeta a granularidade e a eficiência da concorrência no fatiamento completo.	131072
`routing-fields`	Campos para rotear documentos para shards específicos no PolarSearch. Separe vários campos com `;`.	Nenhum
`ignore-delete`	Ignora operações de exclusão. Se definido como `true`, o link não envia exclusões para o índice de destino.	false
`sink.force-index-request`	Força escritas de substituição de linha completa no modo Index. Caso contrário, o modo Update aplica atualizações parciais de campos.	false
`sink.json-flatten.fields`	Campos que exigem análise de JSON. Separe vários campos com `;`.	Nenhum
`sink.json-flatten.mode`	Modo de análise de JSON: `nested` (padrão, converte em objetos aninhados) ou `flatten` (achata pelo primeiro nível).	nested

Melhores práticas

Conversão automática de campos JSON

Por padrão, a sincronização armazena campos JSON do MySQL como strings no PolarSearch. As search views do PolarDB analisam automaticamente esses campos durante a sincronização, gerando campos aninhados ou achatando o JSON.

Preparação de dados

Execute as instruções SQL a seguir no cluster para criar um banco de dados e uma tabela de exemplo e inserir dados de teste:

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"}}');

Configuração padrão (armazenamento como string)

Search view

CREATE SEARCH VIEW json_test AS SELECT * FROM t1;

ETL stored procedure

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`;
");

Verificar dados

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

Converter c1 para o tipo nested

Search view

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

ETL stored procedure

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'   -- Separate multiple fields with ;
);

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

Verificar dados

{
  "_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" }
    }
  }
}

Achatar c1 pelo primeiro nível JSON

Search view

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

ETL stored procedure

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`;
");

Verificar dados: Após o achatamento, os campos de primeiro nível do JSON tornam-se campos independentes de nível superior no índice do 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" : [ ]
  }
}

Definir campos de roteamento de busca

Especifique um ou mais campos na search view para rotear linhas de dados para shards específicos no PolarSearch.

Preparação de dados

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);

Configurar campos de roteamento

Search view

CREATE SEARCH VIEW routing_test
WITH (
  'routing-fields' = 'c1'   -- Separate multiple fields with ;
) AS SELECT * FROM t1;

ETL stored procedure

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'   -- Separate multiple fields with ;
);

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

Verificar dados: O sistema roteia documentos para o shard correspondente com base no valor do campo c1. O campo _routing registra o valor usado no roteamento.

{ "_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 } }

Ignorar exclusões

Em search views que agregam várias tabelas, o AutoETL atualiza o índice de destino excluindo e reinserindo dados. Para evitar que consultas acessem estados intermediários de dados excluídos, ative o ignore-delete e faça o link ignorar operações de exclusão durante a sincronização.

Search view

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

ETL stored procedure

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`;
");

Após essa configuração, o link deixa de executar operações de exclusão.

Nota

Como os dados não são limpos, o índice do PolarSearch pode aumentar de tamanho. Recomendamos usar um campo na tabela de origem do MySQL para marcar linhas excluídas. Após a sincronização com o PolarSearch, utilize uma tarefa agendada para limpar os documentos marcados.

Escritas de substituição

Ao escrever em um índice do PolarSearch com documento existente, o AutoETL usa o modo Update por padrão e atualiza apenas os campos da search view. Para cenários que exigem substituição completa da linha, configure sink.force-index-request para ativar o modo de escrita Index.

Search view

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

ETL stored procedure

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`;
");