SeaTunnel + Elasticsearch 数据入仓实战：从批量导入到 CDC 的 4 个同步场景完整指南

【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel

用 SeaTunnel 的 Elasticsearch 连接器，只写一份 HOCON 配置，就能跑通 MySQL CDC 实时入仓、批量导入、多表动态路由、向量入仓这 4 类数据同步任务，全程不写一行自定义代码，本地环境 20 分钟左右可验证效果。

这个连接器对 ES 2.x 到 8.x 的全版本兼容，CDC 变更捕获、批量写入、HTTPS 加密等常用能力开箱即用，是项目中使用频率最高的 Sink 插件之一。

快速上手：第一次导入 ES 的最小配置

环境要求

• JDK 8 及以上
• SeaTunnel 2.x 发行版
• 可访问的 Elasticsearch 集群（2.x ~ 8.x）

最小配置，新建first_es.conf：

source { Jdbc { url = "jdbc:mysql://localhost:3306/ecommerce" user = "root" password = "123456" query = "SELECT * FROM order_info" } } sink { Elasticsearch { hosts = ["localhost:9200"] index = "order_info" schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST" data_save_mode = "APPEND_DATA" } }

提交任务：

bin/seatunnel.sh --config first_es.conf

任务结束后用 curl 或 Kibana 查一下order_info索引，能命中数据即说明链路已跑通。

• hosts：ES 的 HTTP 地址，列表形式，可填多个节点
• index：目标索引名，支持${字段名}变量写法
• schema_save_mode：目标索引已存在时的处理策略，这里表示不存在就创建、已存在就跳过
• data_save_mode：APPEND_DATA追加写入，另有DROP_DATA（先清空）和ERROR_WHEN_DATA_EXISTS（有数据即报错）

实战场景一：存量数据批量导入

适用场景：把关系库中的大表一次性搬进 ES，供搜索或报表消费。

最小配置

source { Jdbc { url = "jdbc:mysql://localhost:3306/ecommerce" user = "root" password = "123456" query = "SELECT * FROM order_info" } } sink { Elasticsearch { hosts = ["localhost:9200"] index = "order_info_full" schema_save_mode = "RECREATE_SCHEMA" data_save_mode = "DROP_DATA" max_batch_size = 2000 } }

关键参数：

• RECREATE_SCHEMA+DROP_DATA组合实现索引和数据"推倒重建"，适合周期性全量刷新
• max_batch_size是 bulk 批量大小，默认只有 10，大表建议从 1000 起步逐步上调
• 表有唯一键时配上primary_keys，任务重跑不会产生重复文档

实战场景二：CDC 同步最小配置

适用场景：数据库的增删改要实时反映到 ES，典型如订单状态变更后搜索侧立即可查。

最小配置

env { parallelism = 2 job.mode = "STREAMING" checkpoint.interval = 5000 } source { MySQL-CDC { server-id = 5652-5657 username = "st_user" password = "st_pwd" table-names = ["shop.orders"] url = "jdbc:mysql://localhost:3306/shop" } } sink { Elasticsearch { hosts = ["localhost:9200"] index = "orders_cdc" schema_save_mode = "IGNORE" primary_keys = ["order_id"] max_batch_size = 1000 max_retry_count = 3 } }

关键参数：

• primary_keys是 CDC 必配项，用来生成 ES 文档_id；我的经验是这一步最容易漏，漏了之后失败重试会直接产生重复文档
• checkpoint.interval单位为毫秒，控制检查点频率，故障恢复时数据不丢的关键
• Zeta 引擎下再配sink.flush.interval，让没攒够批量阈值的请求也能定时刷出，降低端到端延迟

实战场景三：多表动态路由到不同索引

适用场景：一个任务同步同一库的多张表，每张表自动落到对应索引，不用拆成多份配置。

最小配置

source { MySQL-CDC { server-id = 5652-5657 username = "st_user" password = "st_pwd" table-names = ["shop.orders", "shop.products"] url = "jdbc:mysql://localhost:3306/shop" } } sink { Elasticsearch { hosts = ["localhost:9200"] index = "${table_name}" schema_save_mode = "IGNORE" primary_keys = ["${primary_key}"] } }

关键参数：

• index中的${table_name}按运行期行数据取表名，自动完成"表到索引"的路由
• ${primary_key}同理取每张表自己的主键，保证所有表的_id生成逻辑一致
• 单表写入压力大时，可用multi_table_sink_replica调高该表的写入副本数

进阶：向量字段入仓

做语义搜索、推荐这类场景时，embedding 向量需要以 dense_vector 形式落到 ES。连接器原生支持，要求 ES 7.3 及以上：

sink { Elasticsearch { hosts = ["localhost:9200"] index = "product_reviews" schema_save_mode = "IGNORE" vectorization_fields = ["review_embedding"] vector_dimensions = 1024 } }

• vectorization_fields声明哪些字段是 embedding 向量（上游值为 ByteBuffer 二进制）
• vector_dimensions给出向量维度，ES 据此把字段映射为固定维度向量

生产环境检查清单：批量写入调优与 SSL 安全连接

上线前逐项核对下表即可：

自签名证书或主机名不匹配时的临时写法：

sink { Elasticsearch { hosts = ["https://es-intranet:9200"] username = "elastic" password = "xxx" tls_verify_certificate = false } }

⚠️tls_verify_certificate = false只建议内网测试环境使用，公网链路务必保持证书校验。

Zeta 引擎的 Web 控制台可以持续查看同步任务的运行状态与 Worker 资源占用：

常见问题自诊表

写在最后

从批量导入到 CDC 实时同步，SeaTunnel 的 Elasticsearch 连接器都只需要一份配置文件：批量、多表路由、向量入仓，都是"改参数、跑任务、验结果"三步。文中所有参数均在连接器选项类中定义，你可以按需自由组合扩展。

参考资料

• 官方文档（中文）：docs/zh/connectors/sink/Elasticsearch.md
• 连接器源码目录：seatunnel-connectors-v2/connector-elasticsearch/
• Sink 参数定义：ElasticsearchSinkOptions.java、ElasticsearchBaseOptions.java
• JVM 参数模板：config/jvm_options

【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel