This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 81b2b2c46d [Docs] Add MySQL CDC to Elasticsearch scenario recipe
(#11480)
81b2b2c46d is described below
commit 81b2b2c46d92fd444fa973d1ca59c240973e79f8
Author: Daniel <[email protected]>
AuthorDate: Sun Jul 19 16:55:47 2026 +0800
[Docs] Add MySQL CDC to Elasticsearch scenario recipe (#11480)
Co-authored-by: DanielLeens <[email protected]>
Co-authored-by: David Zollo <[email protected]>
---
.../recipes/mysql-cdc-to-elasticsearch.md | 228 +++++++++++++++
docs/en/getting-started/recipes/overview.md | 11 +-
docs/sidebars.js | 1 +
.../recipes/mysql-cdc-to-elasticsearch.md | 228 +++++++++++++++
docs/zh/getting-started/recipes/overview.md | 19 +-
.../connector-elasticsearch-e2e/pom.xml | 6 +
.../elasticsearch/ElasticsearchRecipeIT.java | 323 +++++++++++++++++++++
.../resources/ddl/recipe_crm_customer_profile.sql | 32 ++
.../mysqlcdc_to_elasticsearch_recipe.conf | 77 +++++
9 files changed, 911 insertions(+), 14 deletions(-)
diff --git a/docs/en/getting-started/recipes/mysql-cdc-to-elasticsearch.md
b/docs/en/getting-started/recipes/mysql-cdc-to-elasticsearch.md
new file mode 100644
index 0000000000..790a485f67
--- /dev/null
+++ b/docs/en/getting-started/recipes/mysql-cdc-to-elasticsearch.md
@@ -0,0 +1,228 @@
+---
+title: MySQL CDC to Elasticsearch with filtering, conversion, and custom fields
+---
+
+# MySQL CDC to Elasticsearch with filtering, cleanup, and custom fields
+
+This recipe continuously synchronizes customer profiles from MySQL to
Elasticsearch, filters unmanaged rows, normalizes phone numbers, maps status
values, and adds source metadata.
+
+The pipeline is:
+
+- `MySQL-CDC` reads `crm.customer_profile`
+- `Metadata` exposes database, table, and row kind
+- `Replace` removes `-` from the phone number
+- `Sql` filters rows and derives `status_name` plus `sync_source`
+- `Elasticsearch` writes documents by primary key
+
+## Prerequisites
+
+1. Finish [Run your first job](../locally/run-your-first-job.md) and make sure
local execution works.
+
+2. Install the connectors required by this recipe. Follow [Deployment >
Download The Connector
Plugins](../locally/deployment.md#download-the-connector-plugins), then keep
these entries in `config/plugin_config`:
+
+```plugin_config
+--seatunnel-connectors--
+connector-cdc-mysql
+connector-elasticsearch
+--end--
+```
+
+```bash
+cd "${SEATUNNEL_HOME}"
+sh bin/install-plugin.sh
+ls connectors | grep -E 'connector-(cdc-mysql|elasticsearch)'
+```
+
+3. For the Zeta engine, place MySQL Connector/J in `${SEATUNNEL_HOME}/lib` and
verify that the JAR is visible:
+
+```bash
+ls "${SEATUNNEL_HOME}/lib" | grep 'mysql-connector'
+```
+
+4. Create the CDC account without granting schema-changing privileges:
+
+```sql
+CREATE USER IF NOT EXISTS 'st_user_source'@'%' IDENTIFIED BY 'mysqlpw';
+GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT,
LOCK TABLES
+ON *.* TO 'st_user_source'@'%';
+FLUSH PRIVILEGES;
+```
+
+5. Verify that MySQL binlog is ready for CDC:
+
+```sql
+SHOW VARIABLES WHERE variable_name IN ('log_bin', 'binlog_format',
'binlog_row_image');
+```
+
+The expected values are `log_bin = ON`, `binlog_format = ROW`, and
`binlog_row_image = FULL`.
+
+6. Make Elasticsearch reachable with HTTPS authentication. This example uses
Elasticsearch 8.9.0. Verify connectivity with an address and account that
SeaTunnel can use:
+
+```bash
+curl --cacert /path/to/http_ca.crt \
+ -u elastic:your_password https://elasticsearch.example.com:9200
+```
+
+Copy the trusted CA certificate to every SeaTunnel node and use that local
path for `tls_truststore_path` in the job configuration.
+
+## Prepare the source data
+
+Create the source table and load two initial rows:
+
+```sql
+CREATE DATABASE IF NOT EXISTS crm;
+USE crm;
+
+CREATE TABLE customer_profile (
+ id BIGINT NOT NULL PRIMARY KEY,
+ name VARCHAR(64) NOT NULL,
+ phone VARCHAR(32) NOT NULL,
+ email VARCHAR(128) NOT NULL,
+ status INT NOT NULL,
+ city VARCHAR(64) NOT NULL
+);
+
+INSERT INTO customer_profile (id, name, phone, email, status, city) VALUES
+ (1001, ' Alice Zhang ', '138-0000-1111', '[email protected]', 1, 'Shanghai'),
+ (900, 'Bob Li', '139-8888-2222', '[email protected]', 0, 'Beijing');
+```
+
+After the job starts and the initial snapshot is visible in Elasticsearch,
execute these incremental changes:
+
+```sql
+UPDATE crm.customer_profile
+SET name = ' Alice Zhang ', phone = '138-9999-0000', status = 2
+WHERE id = 1001;
+
+INSERT INTO crm.customer_profile (id, name, phone, email, status, city) VALUES
+ (1003, 'Carol Wang', '137-1234-8888', '[email protected]', 1, 'Hangzhou');
+```
+
+## Complete job configuration
+
+Replace the example hostnames and credentials below with values reachable from
the SeaTunnel process. Each concurrently running MySQL CDC job must use a
unique `server-id` range.
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
+source {
+ MySQL-CDC {
+ plugin_output = "mysql_customer_raw"
+ url = "jdbc:mysql://mysql.example.com:3306/crm"
+ username = "st_user_source"
+ password = "mysqlpw"
+ server-id = 5701-5704
+ table-names = ["crm.customer_profile"]
+ startup.mode = "initial"
+ schema-changes.enabled = false
+ }
+}
+
+transform {
+ Metadata {
+ plugin_input = "mysql_customer_raw"
+ plugin_output = "mysql_customer_with_meta"
+ metadata_fields {
+ Database = source_database
+ Table = source_table
+ RowKind = row_kind
+ }
+ }
+
+ Replace {
+ plugin_input = "mysql_customer_with_meta"
+ plugin_output = "mysql_customer_cleaned"
+ replace_fields = ["phone"]
+ pattern = "-"
+ replacement = ""
+ is_regex = false
+ }
+
+ Sql {
+ plugin_input = "mysql_customer_cleaned"
+ plugin_output = "es_customer_profile"
+ query = "select id, trim(name) as name, phone, email, city, case when
status = 1 then 'ACTIVE' when status = 2 then 'FROZEN' else 'OTHER' end as
status_name, source_database, source_table, row_kind, 'mysql_cdc' as
sync_source from dual where id >= 1000"
+ }
+}
+
+sink {
+ Elasticsearch {
+ plugin_input = "es_customer_profile"
+ hosts = ["https://elasticsearch.example.com:9200";]
+ username = "elastic"
+ password = "elasticsearch"
+ tls_verify_certificate = true
+ tls_verify_hostname = true
+ tls_truststore_path = "/path/to/http_ca.crt"
+ index = "recipe_customer_profile"
+ primary_keys = ["id"]
+ max_batch_size = 1
+ schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
+ data_save_mode = "APPEND_DATA"
+ }
+}
+```
+
+Save the adapted config as
`${SEATUNNEL_HOME}/config/mysql-cdc-to-elasticsearch.conf`, then submit it with
the Zeta local engine:
+
+```bash
+cd "${SEATUNNEL_HOME}"
+./bin/seatunnel.sh --config ./config/mysql-cdc-to-elasticsearch.conf -m local
+```
+
+In another terminal, query the indexed documents before and after executing
the incremental SQL above:
+
+```bash
+curl --cacert /path/to/http_ca.crt -u elastic:your_password \
+
'https://elasticsearch.example.com:9200/recipe_customer_profile/_search?pretty&sort=id'
+```
+
+## Verify the result
+
+Confirm the following results in the Elasticsearch response:
+
+- Only one document existed after the snapshot phase.
+- Document `1001` was kept because its immutable primary key is in the managed
range `id >= 1000`.
+- Row `900` was filtered out because it is outside that primary-key range.
+- `trim(name)` changed `' Alice Zhang '` into `Alice Zhang`.
+- `Replace` changed `138-0000-1111` into `13800001111`.
+- The indexed document contained `status_name=ACTIVE`, `source_database=crm`,
`source_table=customer_profile`, and `sync_source=mysql_cdc`.
+- After the update, document `1001` became `status_name=FROZEN`,
`phone=13899990000`, and `row_kind=+U`.
+- After inserting `1003`, Elasticsearch contained a second document with
`status_name=ACTIVE`, `phone=13712348888`, `row_kind=+I`, and
`sync_source=mysql_cdc`.
+
+## Why each step exists in this recipe
+
+### 1. `Replace` is used for lightweight field normalization
+
+This job uses `Replace` only for one purpose: remove `-` from `phone` before
indexing.
+
+### 2. `Sql` owns the business filter
+
+The SQL contains the full business rule:
+
+- keep only the managed primary-key range `id >= 1000`
+- map `status` to `ACTIVE` or `FROZEN`
+- add constant field `sync_source`
+
+The filter intentionally uses the immutable primary key. Avoid filtering CDC
updates by mutable fields such as `is_deleted` or `status` before an upsert
sink: when a previously indexed row stops matching, no delete event is emitted
for the old Elasticsearch document. For soft deletion, keep the deletion flag
in Elasticsearch and filter it at query time, or use a pipeline that explicitly
converts the transition into a delete event.
+
+### 3. Elasticsearch write settings
+
+The sink uses these settings together with the expected results above:
+
+- `primary_keys = ["id"]`
+- `max_batch_size = 1`
+- `tls_verify_certificate = true` and `tls_verify_hostname = true`
+- `schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"`
+- `data_save_mode = "APPEND_DATA"`
+
+## Related docs
+
+- [`MySQL-CDC` source connector](../../connectors/source/MySQL-CDC.md)
+- [`Elasticsearch` sink connector](../../connectors/sink/Elasticsearch.md)
+- [`Metadata` transform](../../transforms/metadata.md)
+- [`Replace` transform](../../transforms/replace.md)
+- [`Sql` transform](../../transforms/sql.md)
diff --git a/docs/en/getting-started/recipes/overview.md
b/docs/en/getting-started/recipes/overview.md
index 25dc1282e3..926f72837c 100644
--- a/docs/en/getting-started/recipes/overview.md
+++ b/docs/en/getting-started/recipes/overview.md
@@ -12,14 +12,15 @@ The recipes in this section include concrete prerequisites,
complete configurati
| Goal | Start here |
| --- | --- |
-| CDC from MySQL into an analytics database | [MySQL CDC to
Doris](./mysql-cdc-to-doris.md) |
| CDC from MySQL into Kafka with metadata headers | [MySQL CDC to
Kafka](./mysql-cdc-to-kafka.md) |
+| CDC from MySQL into Elasticsearch with filtering and field shaping | [MySQL
CDC to Elasticsearch](./mysql-cdc-to-elasticsearch.md) |
| Batch migration between relational databases with row transformation | [JDBC
to JDBC](./jdbc-to-jdbc.md) |
| JDBC extraction into object storage | [JDBC to S3](./jdbc-to-s3.md) |
-| Streaming from Kafka into a table format | [Kafka to
Iceberg](./kafka-to-iceberg.md) |
-| CDC from PostgreSQL into a table format | [PostgreSQL CDC to
Iceberg](./postgresql-cdc-to-iceberg.md) |
-| HTTP ingestion into a relational target | [HTTP to JDBC](./http-to-jdbc.md) |
-| File-based loading into an analytical system | [File to
StarRocks](./file-to-starrocks.md) |
+| Streaming from Kafka into Iceberg | [Kafka to
Iceberg](./kafka-to-iceberg.md) |
+| CDC from PostgreSQL into Iceberg | [PostgreSQL CDC to
Iceberg](./postgresql-cdc-to-iceberg.md) |
+| HTTP ingestion into JDBC | [HTTP to JDBC](./http-to-jdbc.md) |
+| CDC from MySQL into Doris | [MySQL CDC to Doris](./mysql-cdc-to-doris.md) |
+| File-based loading into StarRocks | [File to
StarRocks](./file-to-starrocks.md) |
| Multi-table CDC orchestration | [Multi-Table CDC](./multi-table-cdc.md) |
## How To Read A Recipe
diff --git a/docs/sidebars.js b/docs/sidebars.js
index 9ec30ba53a..d3adbdd44b 100644
--- a/docs/sidebars.js
+++ b/docs/sidebars.js
@@ -162,6 +162,7 @@ const sidebars = {
"items": [
"getting-started/recipes/mysql-cdc-to-doris",
"getting-started/recipes/mysql-cdc-to-kafka",
+ "getting-started/recipes/mysql-cdc-to-elasticsearch",
"getting-started/recipes/jdbc-to-s3",
"getting-started/recipes/kafka-to-iceberg",
"getting-started/recipes/postgresql-cdc-to-iceberg",
diff --git a/docs/zh/getting-started/recipes/mysql-cdc-to-elasticsearch.md
b/docs/zh/getting-started/recipes/mysql-cdc-to-elasticsearch.md
new file mode 100644
index 0000000000..8fcf33fe94
--- /dev/null
+++ b/docs/zh/getting-started/recipes/mysql-cdc-to-elasticsearch.md
@@ -0,0 +1,228 @@
+---
+title: MySQL CDC 到 Elasticsearch：过滤、转换与自定义字段
+---
+
+# MySQL CDC 到 Elasticsearch：过滤、清洗与自定义补字段
+
+本教程把客户资料从 MySQL 持续同步到 Elasticsearch，同时过滤非管理范围的数据、清洗手机号、映射状态值并补充来源元数据。
+
+完整链路是：
+
+- `MySQL-CDC` 读取 `crm.customer_profile`
+- `Metadata` 暴露库名、表名和行变更类型
+- `Replace` 去掉手机号中的 `-`
+- `Sql` 负责行过滤、状态映射和补充 `sync_source`
+- `Elasticsearch` 按主键把数据写入索引
+
+## 前置条件
+
+1. 先完成 [跑第一个任务](../locally/run-your-first-job.md)，确认本地执行链路正常。
+
+2. 按照 [部署 > 下载连接器插件](../locally/deployment.md#下载连接器插件) 安装本教程需要的连接器，并在
`config/plugin_config` 中保留下面两项：
+
+```plugin_config
+--seatunnel-connectors--
+connector-cdc-mysql
+connector-elasticsearch
+--end--
+```
+
+```bash
+cd "${SEATUNNEL_HOME}"
+sh bin/install-plugin.sh
+ls connectors | grep -E 'connector-(cdc-mysql|elasticsearch)'
+```
+
+3. 使用 Zeta 引擎时，把 MySQL Connector/J 放入 `${SEATUNNEL_HOME}/lib`，并确认 JAR 已经落盘：
+
+```bash
+ls "${SEATUNNEL_HOME}/lib" | grep 'mysql-connector'
+```
+
+4. 创建 CDC 账号，不授予修改表结构的权限：
+
+```sql
+CREATE USER IF NOT EXISTS 'st_user_source'@'%' IDENTIFIED BY 'mysqlpw';
+GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT,
LOCK TABLES
+ON *.* TO 'st_user_source'@'%';
+FLUSH PRIVILEGES;
+```
+
+5. 检查 MySQL binlog 是否满足 CDC 要求：
+
+```sql
+SHOW VARIABLES WHERE variable_name IN ('log_bin', 'binlog_format',
'binlog_row_image');
+```
+
+期望值是 `log_bin = ON`、`binlog_format = ROW`、`binlog_row_image = FULL`。
+
+6. 确保 Elasticsearch 可以通过 HTTPS 认证访问。本示例使用 Elasticsearch 8.9.0。请使用 SeaTunnel
能访问的地址和账号检查连通性：
+
+```bash
+curl --cacert /path/to/http_ca.crt \
+ -u elastic:your_password https://elasticsearch.example.com:9200
+```
+
+把可信 CA 证书复制到每个 SeaTunnel 节点，并在任务配置的 `tls_truststore_path` 中填写对应的本地路径。
+
+## 准备源表数据
+
+创建源表并插入两条初始数据：
+
+```sql
+CREATE DATABASE IF NOT EXISTS crm;
+USE crm;
+
+CREATE TABLE customer_profile (
+ id BIGINT NOT NULL PRIMARY KEY,
+ name VARCHAR(64) NOT NULL,
+ phone VARCHAR(32) NOT NULL,
+ email VARCHAR(128) NOT NULL,
+ status INT NOT NULL,
+ city VARCHAR(64) NOT NULL
+);
+
+INSERT INTO customer_profile (id, name, phone, email, status, city) VALUES
+ (1001, ' Alice Zhang ', '138-0000-1111', '[email protected]', 1, 'Shanghai'),
+ (900, 'Bob Li', '139-8888-2222', '[email protected]', 0, 'Beijing');
+```
+
+任务启动且初始快照已经可以在 Elasticsearch 中查询到后，执行下面两条增量变更：
+
+```sql
+UPDATE crm.customer_profile
+SET name = ' Alice Zhang ', phone = '138-9999-0000', status = 2
+WHERE id = 1001;
+
+INSERT INTO crm.customer_profile (id, name, phone, email, status, city) VALUES
+ (1003, 'Carol Wang', '137-1234-8888', '[email protected]', 1, 'Hangzhou');
+```
+
+## 完整任务配置
+
+请把下面的示例主机名和凭据替换为 SeaTunnel 进程可以访问的实际值。并发运行的 MySQL CDC 任务必须使用不同的 `server-id` 范围。
+
+```hocon
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
+source {
+ MySQL-CDC {
+ plugin_output = "mysql_customer_raw"
+ url = "jdbc:mysql://mysql.example.com:3306/crm"
+ username = "st_user_source"
+ password = "mysqlpw"
+ server-id = 5701-5704
+ table-names = ["crm.customer_profile"]
+ startup.mode = "initial"
+ schema-changes.enabled = false
+ }
+}
+
+transform {
+ Metadata {
+ plugin_input = "mysql_customer_raw"
+ plugin_output = "mysql_customer_with_meta"
+ metadata_fields {
+ Database = source_database
+ Table = source_table
+ RowKind = row_kind
+ }
+ }
+
+ Replace {
+ plugin_input = "mysql_customer_with_meta"
+ plugin_output = "mysql_customer_cleaned"
+ replace_fields = ["phone"]
+ pattern = "-"
+ replacement = ""
+ is_regex = false
+ }
+
+ Sql {
+ plugin_input = "mysql_customer_cleaned"
+ plugin_output = "es_customer_profile"
+ query = "select id, trim(name) as name, phone, email, city, case when
status = 1 then 'ACTIVE' when status = 2 then 'FROZEN' else 'OTHER' end as
status_name, source_database, source_table, row_kind, 'mysql_cdc' as
sync_source from dual where id >= 1000"
+ }
+}
+
+sink {
+ Elasticsearch {
+ plugin_input = "es_customer_profile"
+ hosts = ["https://elasticsearch.example.com:9200";]
+ username = "elastic"
+ password = "elasticsearch"
+ tls_verify_certificate = true
+ tls_verify_hostname = true
+ tls_truststore_path = "/path/to/http_ca.crt"
+ index = "recipe_customer_profile"
+ primary_keys = ["id"]
+ max_batch_size = 1
+ schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
+ data_save_mode = "APPEND_DATA"
+ }
+}
+```
+
+把调整后的配置保存为 `${SEATUNNEL_HOME}/config/mysql-cdc-to-elasticsearch.conf`，然后使用
Zeta local 模式提交：
+
+```bash
+cd "${SEATUNNEL_HOME}"
+./bin/seatunnel.sh --config ./config/mysql-cdc-to-elasticsearch.conf -m local
+```
+
+在另一个终端中，分别在执行上面的增量 SQL 前后查询索引：
+
+```bash
+curl --cacert /path/to/http_ca.crt -u elastic:your_password \
+
'https://elasticsearch.example.com:9200/recipe_customer_profile/_search?pretty&sort=id'
+```
+
+## 验证结果
+
+在 Elasticsearch 返回结果中确认以下内容：
+
+- 快照阶段最终只写入了 1 条文档。
+- `1001` 被保留下来，因为它的不可变主键属于 `id >= 1000` 的管理范围。
+- `900` 被过滤掉，因为它不在这个主键范围内。
+- `trim(name)` 把 `' Alice Zhang '` 变成了 `Alice Zhang`。
+- `Replace` 把 `138-0000-1111` 变成了 `13800001111`。
+- 初始写入文档里包含
`status_name=ACTIVE`、`source_database=crm`、`source_table=customer_profile`、`sync_source=mysql_cdc`。
+- 更新 `1001` 后，ES 中这条文档变成
`status_name=FROZEN`、`phone=13899990000`、`row_kind=+U`。
+- 插入 `1003` 后，ES 中新增第二条文档，字段包括
`status_name=ACTIVE`、`phone=13712348888`、`row_kind=+I`、`sync_source=mysql_cdc`。
+
+## 为什么这条链路这样写
+
+### 1. `Replace` 负责轻量清洗
+
+这份配置里，`Replace` 只做一件事：把手机号里的 `-` 去掉。
+
+### 2. `Sql` 负责业务过滤与补字段
+
+SQL 包含以下业务规则：
+
+- 只保留 `id >= 1000` 的主键范围
+- 把 `status` 映射成 `ACTIVE` 或 `FROZEN`
+- 新增常量字段 `sync_source`
+
+这里特意使用不可变主键做过滤。不要在写入 upsert sink 之前直接用 `is_deleted`、`status` 这类可变字段过滤 CDC
更新：已经写入 ES 的记录如果后来不再满足条件，旧文档不会自动收到删除事件。软删除场景应把删除标记同步到 Elasticsearch
后在查询时过滤，或者使用能把状态变化明确转换为删除事件的链路。
+
+### 3. Elasticsearch 写入配置
+
+Sink 使用下面这些参数，并产生上述预期结果：
+
+- `primary_keys = ["id"]`
+- `max_batch_size = 1`
+- `tls_verify_certificate = true` 和 `tls_verify_hostname = true`
+- `schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"`
+- `data_save_mode = "APPEND_DATA"`
+
+## 相关文档
+
+- [`MySQL-CDC` source 连接器](../../connectors/source/MySQL-CDC.md)
+- [`Elasticsearch` sink 连接器](../../connectors/sink/Elasticsearch.md)
+- [`Metadata` transform](../../transforms/metadata.md)
+- [`Replace` transform](../../transforms/replace.md)
+- [`Sql` transform](../../transforms/sql.md)
diff --git a/docs/zh/getting-started/recipes/overview.md
b/docs/zh/getting-started/recipes/overview.md
index ed09d7ce58..24570dc14f 100644
--- a/docs/zh/getting-started/recipes/overview.md
+++ b/docs/zh/getting-started/recipes/overview.md
@@ -12,15 +12,16 @@ slug: /getting-started/recipes
| 目标 | 推荐入口 |
| --- | --- |
-| 从 MySQL CDC 同步到分析型数据库 | [MySQL CDC 到 Doris](./mysql-cdc-to-doris.md) |
-| 从 MySQL CDC 同步到 Kafka，并附带元数据消息头 | [MySQL CDC 到
Kafka](./mysql-cdc-to-kafka.md) |
-| 在关系型数据库之间批量迁移，并进行数据过滤和转换 | [JDBC 到 JDBC](./jdbc-to-jdbc.md) |
-| 把 JDBC 数据抽取到对象存储 | [JDBC 到 S3](./jdbc-to-s3.md) |
-| 从 Kafka 流式写入表格式存储 | [Kafka 到 Iceberg](./kafka-to-iceberg.md) |
-| 从 PostgreSQL CDC 同步到表格式存储 | [PostgreSQL CDC 到
Iceberg](./postgresql-cdc-to-iceberg.md) |
-| 把 HTTP 数据写入关系型数据库 | [HTTP 到 JDBC](./http-to-jdbc.md) |
-| 把文件数据加载到分析型系统 | [文件到 StarRocks](./file-to-starrocks.md) |
-| 多表 CDC 编排 | [多表 CDC](./multi-table-cdc.md) |
+| 从 MySQL CDC 数据实时同步到 Kafka 并附带元数据消息头 | [MySQL CDC 到
Kafka](./mysql-cdc-to-kafka.md) |
+| 从 MySQL CDC 数据实时同步到 Elasticsearch 并完成过滤与字段整形 | [MySQL CDC 到
Elasticsearch](./mysql-cdc-to-elasticsearch.md) |
+| 从 JDBC 数据批量同步到 JDBC 并进行数据过滤和转换 | [JDBC 到 JDBC](./jdbc-to-jdbc.md) |
+| 从 JDBC 数据批量同步到 S3 对象存储 | [JDBC 到 S3](./jdbc-to-s3.md) |
+| 从 Kafka 流式写入 Iceberg | [Kafka 到 Iceberg](./kafka-to-iceberg.md) |
+| 从 PostgreSQL CDC 数据实时同步到 Iceberg | [PostgreSQL CDC 到
Iceberg](./postgresql-cdc-to-iceberg.md) |
+| 从 HTTP 数据批量写入 JDBC 关系型数据库 | [HTTP 到 JDBC](./http-to-jdbc.md) |
+| 从 MySQL CDC 数据实时同步到 Doris | [MySQL CDC 到 Doris](./mysql-cdc-to-doris.md) |
+| 从 File 文件数据同步到 StarRocks| [文件到 StarRocks](./file-to-starrocks.md) |
+| CDC 实时多表数据同步 | [多表 CDC](./multi-table-cdc.md) |
## 阅读示例时建议这样看
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/pom.xml
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/pom.xml
index 863a8b205f..757e0ed2d1 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/pom.xml
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/pom.xml
@@ -83,5 +83,11 @@
<version>${project.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-transforms-v2</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/java/org/apache/seatunnel/e2e/connector/elasticsearch/ElasticsearchRecipeIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/java/org/apache/seatunnel/e2e/connector/elasticsearch/ElasticsearchRecipeIT.java
new file mode 100644
index 0000000000..b4121789b8
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/java/org/apache/seatunnel/e2e/connector/elasticsearch/ElasticsearchRecipeIT.java
@@ -0,0 +1,323 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.e2e.connector.elasticsearch;
+
+import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.JsonNode;
+import
org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ObjectNode;
+
+import org.apache.seatunnel.common.utils.JsonUtils;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.testutils.MySqlContainer;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.testutils.MySqlVersion;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.testutils.UniqueDatabase;
+import org.apache.seatunnel.e2e.common.TestResource;
+import org.apache.seatunnel.e2e.common.TestSuiteBase;
+import org.apache.seatunnel.e2e.common.container.ContainerExtendedFactory;
+import org.apache.seatunnel.e2e.common.container.EngineType;
+import org.apache.seatunnel.e2e.common.container.TestContainer;
+import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
+import org.apache.seatunnel.e2e.common.junit.TestContainerExtension;
+import org.apache.seatunnel.e2e.common.util.JobIdGenerator;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.elasticsearch.ElasticsearchContainer;
+import org.testcontainers.lifecycle.Startables;
+import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.DockerLoggerFactory;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Stream;
+
+import static org.awaitility.Awaitility.await;
+
+/**
+ * Validates the documented MySQL CDC to Elasticsearch recipe with filtering,
normalization,
+ * metadata enrichment, and derived fields.
+ */
+@Slf4j
+@DisabledOnContainer(
+ value = {},
+ type = {EngineType.SPARK, EngineType.FLINK},
+ disabledReason =
+ "This recipe validates a MySQL CDC streaming pipeline and is
kept on the zeta engine to match the getting-started documentation path.")
+public class ElasticsearchRecipeIT extends TestSuiteBase implements
TestResource {
+
+ // MySQL hostname referenced by the documented job config.
+ private static final String MYSQL_HOST = "mysql_cdc_e2e";
+ // Source database consumed by the CDC job.
+ private static final String MYSQL_DATABASE = "crm";
+ // Application user used for incremental source changes.
+ private static final String MYSQL_USER_NAME = "mysqluser";
+ // Shared password for the MySQL test users.
+ private static final String MYSQL_USER_PASSWORD = "mysqlpw";
+ // Administrative MySQL user used to create the CDC account.
+ private static final String MYSQL_ROOT_USER_NAME = "root";
+ // CDC reader account configured inside the documented job.
+ private static final String MYSQL_CDC_USER_NAME = "st_user_source";
+ // Elasticsearch index populated by the documented sink config.
+ private static final String INDEX_NAME = "recipe_customer_profile";
+ // Driver JAR injected into the MySQL-CDC plugin before the job starts.
+ private static final String DRIVER_JAR =
+
"https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar";;
+
+ // Elasticsearch service used by the recipe job.
+ private ElasticsearchContainer elasticsearchContainer;
+ // MySQL service used by the recipe job.
+ private MySqlContainer mysqlContainer;
+ // Helper that materializes the documented source table and seed rows.
+ private UniqueDatabase crmDatabase;
+
+ // Injects the MySQL JDBC driver required by the MySQL-CDC connector.
+ @TestContainerExtension
+ private final ContainerExtendedFactory extendedFactory =
+ container -> {
+ Container.ExecResult extraCommands =
+ container.execInContainer(
+ "bash",
+ "-c",
+ "mkdir -p /tmp/seatunnel/plugins/MySQL-CDC/lib
&& cd /tmp/seatunnel/plugins/MySQL-CDC/lib && wget "
+ + DRIVER_JAR);
+ Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
+ };
+
+ /**
+ * Starts the external systems and prepares the source data used by the
documented recipe.
+ *
+ * <p>The source and sink hostnames must remain aligned with the submitted
HOCON resource.
+ */
+ @BeforeEach
+ @Override
+ public void startUp() {
+ elasticsearchContainer =
+ new ElasticsearchContainer(
+ DockerImageName.parse(
+
"docker.elastic.co/elasticsearch/elasticsearch:8.9.0"))
+ .withNetwork(NETWORK)
+
.withEnv("cluster.routing.allocation.disk.threshold_enabled", "false")
+ .withNetworkAliases("elasticsearch")
+ .withPassword("elasticsearch")
+ .withStartupAttempts(5)
+ .withStartupTimeout(Duration.ofMinutes(5))
+ .withLogConsumer(
+ new Slf4jLogConsumer(
+
DockerLoggerFactory.getLogger("elasticsearch:8.9.0")));
+ mysqlContainer =
+ new MySqlContainer(MySqlVersion.V8_0)
+
.withConfigurationOverride("docker/server-gtids/my.cnf")
+ .withNetwork(NETWORK)
+ .withNetworkAliases(MYSQL_HOST)
+ .withDatabaseName(MYSQL_DATABASE)
+ .withUsername(MYSQL_USER_NAME)
+ .withPassword(MYSQL_USER_PASSWORD)
+ .withLogConsumer(
+ new Slf4jLogConsumer(
+
DockerLoggerFactory.getLogger("mysql-docker-image")));
+ Startables.deepStart(Stream.of(elasticsearchContainer,
mysqlContainer)).join();
+ createCdcUser();
+ crmDatabase = new UniqueDatabase(mysqlContainer, MYSQL_DATABASE);
+
crmDatabase.setTemplateName("recipe_crm_customer_profile").createAndInitialize();
+ }
+
+ /**
+ * Verifies snapshot filtering and field shaping followed by CDC update
and insert events.
+ *
+ * @param container SeaTunnel runtime that executes the documented
streaming job
+ */
+ @TestTemplate
+ public void testMysqlCdcToElasticsearchRecipe(TestContainer container) {
+ String jobId = String.valueOf(JobIdGenerator.newJobId());
+ AtomicReference<Throwable> jobFailure = new AtomicReference<>();
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ container.executeJob(
+
"/elasticsearch/mysqlcdc_to_elasticsearch_recipe.conf", jobId);
+ } catch (Exception e) {
+ jobFailure.set(e);
+ throw new RuntimeException(e);
+ }
+ });
+
+ await().atMost(120, TimeUnit.SECONDS)
+ .pollInterval(3, TimeUnit.SECONDS)
+ .ignoreExceptions()
+ .untilAsserted(
+ () -> {
+ assertNoJobFailure(jobFailure);
+ Map<Integer, JsonNode> docs = fetchIndexDocs();
+ Assertions.assertEquals(1, docs.size());
+ JsonNode doc1001 = docs.get(1001);
+ Assertions.assertNotNull(doc1001);
+ Assertions.assertEquals("Alice Zhang",
doc1001.get("name").asText());
+ Assertions.assertEquals("13800001111",
doc1001.get("phone").asText());
+ Assertions.assertEquals("ACTIVE",
doc1001.get("status_name").asText());
+ Assertions.assertEquals("crm",
doc1001.get("source_database").asText());
+ Assertions.assertEquals(
+ "customer_profile",
doc1001.get("source_table").asText());
+ Assertions.assertEquals(
+ "mysql_cdc",
doc1001.get("sync_source").asText());
+ Assertions.assertFalse(docs.containsKey(900));
+ });
+
+ executeSql(
+ "UPDATE crm.customer_profile SET name = ' Alice Zhang ', phone
= '138-9999-0000', status = 2 WHERE id = 1001",
+ "INSERT INTO crm.customer_profile (id, name, phone, email,
status, city) VALUES "
+ + "(1003, 'Carol Wang', '137-1234-8888',
'[email protected]', 1, 'Hangzhou')");
+
+ await().atMost(120, TimeUnit.SECONDS)
+ .pollInterval(3, TimeUnit.SECONDS)
+ .ignoreExceptions()
+ .untilAsserted(
+ () -> {
+ assertNoJobFailure(jobFailure);
+ Map<Integer, JsonNode> docs = fetchIndexDocs();
+ Assertions.assertEquals(2, docs.size());
+
+ JsonNode doc1001 = docs.get(1001);
+ JsonNode doc1003 = docs.get(1003);
+ Assertions.assertNotNull(doc1001);
+ Assertions.assertNotNull(doc1003);
+
+ Assertions.assertEquals("FROZEN",
doc1001.get("status_name").asText());
+ Assertions.assertEquals("13899990000",
doc1001.get("phone").asText());
+ Assertions.assertEquals("Alice Zhang",
doc1001.get("name").asText());
+ Assertions.assertEquals("+U",
doc1001.get("row_kind").asText());
+
+ Assertions.assertEquals("ACTIVE",
doc1003.get("status_name").asText());
+ Assertions.assertEquals("13712348888",
doc1003.get("phone").asText());
+ Assertions.assertEquals("+I",
doc1003.get("row_kind").asText());
+ Assertions.assertEquals(
+ "mysql_cdc",
doc1003.get("sync_source").asText());
+ });
+
+ assertNoJobFailure(jobFailure);
+ Assertions.assertEquals("RUNNING", container.getJobStatus(jobId));
+ try {
+ Container.ExecResult cancelResult = container.cancelJob(jobId);
+ Assertions.assertEquals(0, cancelResult.getExitCode(),
cancelResult.getStderr());
+ } catch (Exception e) {
+ throw new IllegalStateException("Failed to cancel the recipe
streaming job", e);
+ }
+ }
+
+ /**
+ * Stops the external systems created for the recipe test.
+ *
+ * <p>Each template invocation owns isolated MySQL and Elasticsearch
containers.
+ */
+ @AfterEach
+ @Override
+ public void tearDown() {
+ if (Objects.nonNull(elasticsearchContainer)) {
+ elasticsearchContainer.close();
+ }
+ if (Objects.nonNull(mysqlContainer)) {
+ mysqlContainer.close();
+ }
+ }
+
+ // Executes incremental source changes with the application user.
+ private void executeSql(String... statements) {
+ try (Connection connection =
+ DriverManager.getConnection(
+ mysqlContainer.getJdbcUrl(MYSQL_DATABASE),
+ MYSQL_USER_NAME,
+ MYSQL_USER_PASSWORD);
+ Statement statement = connection.createStatement()) {
+ for (String sql : statements) {
+ statement.execute(sql);
+ }
+ } catch (SQLException e) {
+ throw new IllegalStateException("Failed to execute MySQL SQL", e);
+ }
+ }
+
+ // Creates the CDC account used by the documented source config.
+ private void createCdcUser() {
+ executeAdminSql(
+ "CREATE USER IF NOT EXISTS '"
+ + MYSQL_CDC_USER_NAME
+ + "'@'%' IDENTIFIED BY '"
+ + MYSQL_USER_PASSWORD
+ + "'",
+ "GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE,
REPLICATION CLIENT, LOCK TABLES ON *.* TO '"
+ + MYSQL_CDC_USER_NAME
+ + "'@'%'");
+ }
+
+ // Executes MySQL account setup statements as the container administrator.
+ private void executeAdminSql(String... statements) {
+ try (Connection connection =
+ DriverManager.getConnection(
+ mysqlContainer.getJdbcUrl(MYSQL_DATABASE),
+ MYSQL_ROOT_USER_NAME,
+ MYSQL_USER_PASSWORD);
+ Statement statement = connection.createStatement()) {
+ for (String sql : statements) {
+ statement.execute(sql);
+ }
+ } catch (SQLException e) {
+ throw new IllegalStateException("Failed to execute MySQL admin
SQL", e);
+ }
+ }
+
+ // Reads all indexed documents and maps them by the documented primary key.
+ private Map<Integer, JsonNode> fetchIndexDocs() throws Exception {
+ Container.ExecResult indexCountResult =
+ elasticsearchContainer.execInContainer(
+ "bash",
+ "-c",
+ "curl -k -u elastic:elasticsearch -H
\"Content-Type:application/json\" "
+ + "-d
'{\"from\":0,\"size\":100,\"query\":{\"match_all\":{}}}' "
+ + "https://localhost:9200/";
+ + INDEX_NAME
+ + "/_search");
+ Assertions.assertEquals(0, indexCountResult.getExitCode(),
indexCountResult.getStderr());
+ ObjectNode jsonNode =
JsonUtils.parseObject(indexCountResult.getStdout());
+ JsonNode hits = jsonNode.get("hits").get("hits");
+ Map<Integer, JsonNode> docs = new HashMap<>();
+ for (JsonNode hit : hits) {
+ JsonNode source = hit.get("_source");
+ docs.put(source.get("id").asInt(), source);
+ }
+ return docs;
+ }
+
+ // Surfaces an asynchronous SeaTunnel job failure while Awaitility polls
sink results.
+ private static void assertNoJobFailure(AtomicReference<Throwable>
jobFailure) {
+ if (jobFailure.get() != null) {
+ Assertions.fail(jobFailure.get());
+ }
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/resources/ddl/recipe_crm_customer_profile.sql
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/resources/ddl/recipe_crm_customer_profile.sql
new file mode 100644
index 0000000000..5aa84aa2ae
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/resources/ddl/recipe_crm_customer_profile.sql
@@ -0,0 +1,32 @@
+--
+-- Licensed to the Apache Software Foundation (ASF) under one or more
+-- contributor license agreements. See the NOTICE file distributed with
+-- this work for additional information regarding copyright ownership.
+-- The ASF licenses this file to You under the Apache License, Version 2.0
+-- (the "License"); you may not use this file except in compliance with
+-- the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+--
+
+USE $DBNAME$;
+
+DROP TABLE IF EXISTS customer_profile;
+CREATE TABLE customer_profile (
+ id BIGINT NOT NULL PRIMARY KEY,
+ name VARCHAR(64) NOT NULL,
+ phone VARCHAR(32) NOT NULL,
+ email VARCHAR(128) NOT NULL,
+ status INT NOT NULL,
+ city VARCHAR(64) NOT NULL
+);
+
+INSERT INTO customer_profile (id, name, phone, email, status, city) VALUES
+ (1001, ' Alice Zhang ', '138-0000-1111', '[email protected]', 1, 'Shanghai'),
+ (900, 'Bob Li', '139-8888-2222', '[email protected]', 0, 'Beijing');
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/resources/elasticsearch/mysqlcdc_to_elasticsearch_recipe.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/resources/elasticsearch/mysqlcdc_to_elasticsearch_recipe.conf
new file mode 100644
index 0000000000..2d2b5afe70
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/resources/elasticsearch/mysqlcdc_to_elasticsearch_recipe.conf
@@ -0,0 +1,77 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+}
+
+source {
+ MySQL-CDC {
+ plugin_output = "mysql_customer_raw"
+ url = "jdbc:mysql://mysql_cdc_e2e:3306/crm"
+ username = "st_user_source"
+ password = "mysqlpw"
+ server-id = 5701-5704
+ table-names = ["crm.customer_profile"]
+ startup.mode = "initial"
+ schema-changes.enabled = false
+ }
+}
+
+transform {
+ Metadata {
+ plugin_input = "mysql_customer_raw"
+ plugin_output = "mysql_customer_with_meta"
+ metadata_fields {
+ Database = source_database
+ Table = source_table
+ RowKind = row_kind
+ }
+ }
+
+ Replace {
+ plugin_input = "mysql_customer_with_meta"
+ plugin_output = "mysql_customer_cleaned"
+ replace_fields = ["phone"]
+ pattern = "-"
+ replacement = ""
+ is_regex = false
+ }
+
+ Sql {
+ plugin_input = "mysql_customer_cleaned"
+ plugin_output = "es_customer_profile"
+ query = "select id, trim(name) as name, phone, email, city, case when
status = 1 then 'ACTIVE' when status = 2 then 'FROZEN' else 'OTHER' end as
status_name, source_database, source_table, row_kind, 'mysql_cdc' as
sync_source from dual where id >= 1000"
+ }
+}
+
+sink {
+ Elasticsearch {
+ plugin_input = "es_customer_profile"
+ hosts = ["https://elasticsearch:9200";]
+ username = "elastic"
+ password = "elasticsearch"
+ tls_verify_certificate = false
+ tls_verify_hostname = false
+ index = "recipe_customer_profile"
+ primary_keys = ["id"]
+ max_batch_size = 1
+ schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST"
+ data_save_mode = "APPEND_DATA"
+ }
+}
