AutoETL parameter configuration and best practices - PolarDB - Centro de documentación de Alibaba Cloud

AutoETL ofrece varias opciones de configuración que le permiten ajustar el comportamiento de sincronización según sus escenarios de negocio. Entre ellas se incluyen la conversión de campos JSON, el enrutamiento de búsqueda, el modo de escritura y los recursos de cómputo que utiliza el enlace. Este tema describe la sintaxis de los parámetros para los dos puntos de entrada de configuración (las search views y los procedimientos almacenados ETL) y presenta las mejores prácticas para escenarios comunes.

Los parámetros se configuran en dos puntos de entrada:

Search view: establezca los parámetros directamente en la cláusula WITH (...) del DDL. Estos parámetros solo surten efecto en la search view actual.
Procedimiento almacenado ETL: compatible con la sintaxis de Flink. La configuración del enlace se establece mediante la variable de sesión esl_link_options. Los parámetros de sincronización (lectura de la tabla de origen y escritura en el destino) se escriben directamente en la cláusula WITH (...) del Flink SQL.

Configuración de recursos de cómputo

Un enlace ETL usa uno o más subprocesos de trabajo para transferir los datos. Usted controla los recursos que consume el enlace mediante los siguientes parámetros:

Parámetro	Descripción	Valor predeterminado
`parallelism`	Concurrencia total del enlace ETL actual.	4
`link.tm.cpu`	Número de núcleos de CPU asignados a cada worker.	2
`link.tm.slot`	Concurrencia de tareas admitida por cada worker.	4

Nota

Una mayor concurrencia de workers no siempre es mejor. También debe considerar la capacidad total de CPU de todos los workers.

El enlace de transferencia se mide en CU. La fórmula es CU = parallelism / link.tm.slot * link.tm.cpu + 1. La siguiente configuración de ejemplo corresponde a 5 CU.

Search view

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

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

Lista de parámetros compatibles

A continuación se resumen todos los parámetros que admite AutoETL actualmente.

Parámetros del enlace

Parámetro	Descripción	Valor predeterminado
`parallelism`	Concurrencia total del enlace ETL actual.	4
`link.tm.cpu`	Número de núcleos de CPU asignados a cada worker.	2
`link.tm.slot`	Concurrencia de tareas admitida por cada worker.	4
`link.param.execution.checkpointing.interval`	Intervalo de checkpoint del enlace ETL.	180s

Parámetros de sincronización

Parámetro	Descripción	Valor predeterminado
`gdn-cluster`	Clúster GDN de destino para la sincronización.	El clúster predeterminado de la instancia actual
`sink-name`	Nombre del índice de la search view en PolarSearch. En los procedimientos almacenados, la tabla de destino se especifica mediante `'index'`.	El nombre de la search view
`scan.incremental.snapshot.chunk.size`	Número de filas en cada fragmento durante la fase de escaneo completo. Este parámetro afecta la granularidad y la eficiencia de concurrencia del fraccionamiento completo.	131072
`routing-fields`	Nombres de los campos utilizados para enrutar documentos a shards específicos en PolarSearch. Separe varios campos con `;`.	Ninguna
`ignore-delete`	Indica si se omiten las operaciones de eliminación. Si se establece en `true`, el enlace deja de enviar operaciones de eliminación al índice de destino.	false
`sink.force-index-request`	Indica si se fuerzan escrituras de reemplazo de fila completa en el modo Index. De lo contrario, se usa el modo Update para actualizar campos parciales.	false
`sink.json-flatten.fields`	Nombres de los campos que requieren análisis JSON. Separe varios campos con `;`.	Ninguna
`sink.json-flatten.mode`	Modo de análisis JSON: `nested` (predeterminado, convierte a objetos anidados) o `flatten` (aplana por el primer nivel).	nested

Mejores prácticas

Conversión automática de campos JSON

De forma predeterminada, los campos JSON de MySQL se almacenan como cadenas cuando se sincronizan con PolarSearch. Las search views de PolarDB admiten el análisis automático de estos campos durante la sincronización hacia PolarSearch, lo que permite generar campos anidados o aplanar los campos JSON.

Preparación de datos

Ejecute las siguientes instrucciones SQL en el clúster para crear una base de datos y una tabla de ejemplo, e insertar datos de prueba:

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

Configuración predeterminada (almacenamiento como cadena)

Search view

CREATE SEARCH VIEW json_test AS SELECT * FROM t1;

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

Compruebe los datos

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

Convierta c1 al tipo anidado

Search view

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

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

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

Compruebe los datos

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

Aplane c1 por el primer nivel JSON

Search view

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

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

Compruebe los datos: después de aplanar, los campos del primer nivel de JSON se convierten en campos independientes de nivel superior en el índice de 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" : [ ]
  }
}

Establezca los campos de enrutamiento de búsqueda

Especifique uno o más nombres de campo en la search view para enrutar las filas de datos a shards específicos en PolarSearch.

Preparación de datos

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

Configure los campos de enrutamiento

Search view

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

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

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

Compruebe los datos: los documentos se enrutan al shard correspondiente según el valor del campo c1. El campo _routing registra el valor utilizado para el enrutamiento.

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

Omita las eliminaciones

En las search views que agregan varias tablas, AutoETL actualiza el índice de destino mediante una eliminación seguida de una inserción. Si desea evitar que las consultas accedan a estados intermedios de datos eliminados, habilite ignore-delete para que el enlace omita las operaciones de eliminación durante la sincronización.

Search view

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

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

Después de aplicar esta configuración, el enlace deja de ejecutar operaciones de eliminación.

Nota

Dado que los datos no se limpian, el índice de PolarSearch puede crecer. Se recomienda usar un campo en la tabla de origen de MySQL para marcar las filas eliminadas. Una vez sincronizados los datos con PolarSearch, programe una tarea para limpiar los documentos marcados.

Escrituras de reemplazo

Cuando AutoETL escribe en un índice de PolarSearch y el documento ya existe, utiliza el modo Update de forma predeterminada. Este modo solo actualiza los campos escritos por la search view. Para escenarios que requieren reemplazar la fila completa, configure sink.force-index-request para habilitar el modo de escritura Index.

Search view

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

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