AutoETL parameter configuration and best practices - PolarDB

Parameters can be configured at two entry points:

Search view: Set parameters inline in the WITH (...) clause of the DDL. The parameters take effect only on the current search view.
ETL stored procedure: Compatible with Flink syntax. The link configuration can be set through the session variable esl_link_options, and the synchronization parameters (source table read and destination write) are written directly in the WITH (...) clause of the Flink SQL.

Computing resource configuration

An ETL link uses one or more worker threads to perform the actual data transfer. You can control the resources used by the link through the following parameters:

Parameter	Description	Default value
`parallelism`	The total concurrency of the current ETL link.	4
`link.tm.cpu`	The number of CPU cores allocated to each worker.	2
`link.tm.slot`	The task concurrency supported by each worker.	4

Note

Higher worker concurrency is not always better. You also need to consider the total CPU capacity of all workers.

The transfer link is measured in CU. The CU formula is CU = parallelism / link.tm.slot * link.tm.cpu + 1. The following example configuration corresponds to 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`;
");

Supported parameter list

The following tables summarize all parameters currently supported by AutoETL.

Link parameters

Parameter	Description	Default value
`parallelism`	The total concurrency of the current ETL link.	4
`link.tm.cpu`	The number of CPU cores allocated to each worker.	2
`link.tm.slot`	The task concurrency supported by each worker.	4
`link.param.execution.checkpointing.interval`	The checkpoint interval of the ETL link.	180s

Synchronization parameters

Parameter	Description	Default value
`gdn-cluster`	The destination GDN cluster for synchronization.	The default cluster of the current instance
`sink-name`	The index name of the search view in PolarSearch. For stored procedures, the destination table is specified through `'index'`.	The search view name
`scan.incremental.snapshot.chunk.size`	The number of rows in each chunk during the full scan phase. This parameter affects the granularity and concurrency efficiency of full slicing.	131072
`routing-fields`	The field names used to route documents to specified shards in PolarSearch. Separate multiple fields with `;`.	None
`ignore-delete`	Whether to ignore delete operations. When set to `true`, the link no longer sends delete operations to the destination index.	false
`sink.force-index-request`	Whether to force full-row replacement writes in Index mode. Otherwise, Update mode is used for partial field updates.	false
`sink.json-flatten.fields`	The field names that require JSON parsing. Separate multiple fields with `;`.	None
`sink.json-flatten.mode`	The JSON parsing mode: `nested` (default, converts to nested objects) / `flatten` (flattens by the first level).	nested

Best practices

Automatic JSON field conversion

By default, MySQL JSON fields are stored as strings when synchronized to PolarSearch. PolarDB search views support automatic parsing of MySQL JSON fields during synchronization to PolarSearch, generating nested fields or flattening JSON fields.

Data preparation

Execute the following SQL statements in the cluster to create a sample database and table and insert test data:

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

Default configuration (string storage)

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

Verify data

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

Convert c1 to nested type

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

Verify data

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

Flatten c1 by the first JSON level

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

Verify data: After flattening, the first-level fields of JSON become independent top-level fields in the PolarSearch index.

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

Set Search routing fields

Specify one or more field names in the search view to route data rows to specified shards in PolarSearch.

Data preparation

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 routing fields

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

Verify data: Documents are routed to the corresponding shard based on the value of the c1 field. The _routing field records the value used for 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 } }

Ignore deletes

For search views that aggregate multiple tables, AutoETL updates the destination index by deleting and then inserting. If you do not want queries to access intermediate states of deleted data, you can enable ignore-delete to make the link skip delete operations during synchronization.

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

After this configuration, the link no longer performs delete operations.

Note

Because data is not cleaned up, the PolarSearch index may grow larger. We recommend that you use a field in the MySQL source table to mark deleted rows. After synchronization to PolarSearch, you can use a scheduled task to clean up the marked documents.

Replace writes

When AutoETL writes to a PolarSearch index and the document already exists, it uses Update mode by default, which only updates the fields written by the search view. For scenarios that require full-row replacement, you can configure sink.force-index-request to enable Index write mode.

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