Konfigurasi parameter AutoETL dan praktik terbaik - PolarDB

AutoETL menyediakan berbagai opsi konfigurasi yang memungkinkan Anda menyesuaikan perilaku sinkronisasi sesuai skenario bisnis, seperti konversi field JSON, routing pencarian, mode penulisan, dan sumber daya komputasi yang digunakan oleh link. Topik ini menjelaskan sintaks parameter untuk dua titik masuk konfigurasi—search view dan ETL stored procedure—serta memberikan praktik terbaik untuk skenario umum.

Parameter dapat dikonfigurasi di dua titik masuk:

Search view: Tetapkan parameter secara inline dalam klausa WITH (...) DDL. Parameter hanya berlaku untuk search view tersebut.
ETL stored procedure: Kompatibel dengan sintaks Flink. Konfigurasi link dapat diatur melalui variabel sesi esl_link_options, sedangkan parameter sinkronisasi (pembacaan tabel sumber dan penulisan tujuan) ditentukan langsung dalam klausa WITH (...) Flink SQL.

Computing resource configuration

Link ETL menggunakan satu atau beberapa worker thread untuk melakukan transfer data aktual. Anda dapat mengontrol sumber daya yang digunakan oleh link melalui parameter berikut:

Parameter	Description	Default value
`parallelism`	Total concurrency dari link ETL saat ini.	4
`link.tm.cpu`	Jumlah core CPU yang dialokasikan untuk setiap worker.	2
`link.tm.slot`	Concurrency tugas yang didukung oleh setiap worker.	4

Catatan

Concurrency worker yang lebih tinggi tidak selalu lebih baik. Pertimbangkan juga kapasitas CPU total dari semua worker.

Link transfer diukur dalam CU. Rumus CU adalah CU = parallelism / link.tm.slot * link.tm.cpu + 1. Contoh konfigurasi berikut berkorespondensi dengan 5 CU.

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

Tabel berikut merangkum semua parameter yang saat ini didukung oleh AutoETL.

Link parameters

Parameter	Description	Default value
`parallelism`	Total concurrency dari link ETL saat ini.	4
`link.tm.cpu`	Jumlah core CPU yang dialokasikan untuk setiap worker.	2
`link.tm.slot`	Concurrency tugas yang didukung oleh setiap worker.	4
`link.param.execution.checkpointing.interval`	Interval checkpoint dari link ETL.	180s

Synchronization parameters

Parameter	Description	Default value
`gdn-cluster`	Kluster GDN tujuan untuk sinkronisasi.	Kluster default dari instans saat ini
`sink-name`	Nama indeks dari search view di PolarSearch. Untuk stored procedure, tabel tujuan ditentukan melalui `'index'`.	Nama search view
`scan.incremental.snapshot.chunk.size`	Jumlah baris dalam setiap chunk selama fase full scan. Parameter ini memengaruhi granularitas dan efisiensi konkurensi pemotongan penuh.	131072
`routing-fields`	Nama field yang digunakan untuk merutekan dokumen ke shard tertentu di PolarSearch. Pisahkan beberapa field dengan `;`.	None
`ignore-delete`	Apakah akan mengabaikan operasi delete. Saat diatur ke `true`, link tidak lagi mengirim operasi delete ke indeks tujuan.	false
`sink.force-index-request`	Apakah akan memaksa penulisan penggantian baris penuh dalam mode Index. Jika tidak, mode Update digunakan untuk pembaruan field parsial.	false
`sink.json-flatten.fields`	Nama field yang memerlukan parsing JSON. Pisahkan beberapa field dengan `;`.	None
`sink.json-flatten.mode`	Mode parsing JSON: `nested` (default, mengonversi ke objek bersarang) / `flatten` (meratakan hingga level pertama).	nested

Best practices

Automatic JSON field conversion

Secara default, field JSON MySQL disimpan sebagai string saat disinkronkan ke PolarSearch. Search view PolarDB mendukung parsing otomatis field JSON MySQL selama sinkronisasi ke PolarSearch, sehingga menghasilkan field bersarang atau field JSON yang diratakan.

Data preparation

Jalankan pernyataan SQL berikut di kluster untuk membuat database dan tabel sampel serta memasukkan data uji:

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: Setelah diratakan, field tingkat pertama JSON menjadi field tingkat atas independen dalam indeks 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" : [ ]
  }
}

Set Search routing fields

Tentukan satu atau beberapa nama field dalam search view untuk merutekan baris data ke shard tertentu di 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: Dokumen dirutekan ke shard yang sesuai berdasarkan nilai field c1. Field _routing mencatat nilai yang digunakan untuk 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

Untuk search view yang menggabungkan beberapa tabel, AutoETL memperbarui indeks tujuan dengan menghapus lalu memasukkan kembali. Jika Anda tidak ingin kueri mengakses keadaan antara setelah data dihapus, aktifkan ignore-delete agar link melewati operasi delete selama sinkronisasi.

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

Setelah konfigurasi ini, link tidak lagi melakukan operasi delete.

Catatan

Karena data tidak dibersihkan, ukuran indeks PolarSearch mungkin bertambah. Disarankan untuk menggunakan field di tabel sumber MySQL guna menandai baris yang dihapus. Setelah disinkronkan ke PolarSearch, Anda dapat menggunakan Tugas terjadwal untuk membersihkan dokumen yang ditandai tersebut.

Replace writes

Saat AutoETL menulis ke indeks PolarSearch dan dokumen sudah ada, secara default menggunakan mode Update, yang hanya memperbarui field yang ditulis oleh search view. Untuk skenario yang memerlukan penggantian baris penuh, konfigurasikan sink.force-index-request untuk mengaktifkan mode penulisan 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`;
");