功能概述 #

Elasticsearch CDC Pipeline Sink Connector 用于将火山引擎流式计算 Flink 版 Pipeline 作业捕获的 Insert、Update 和 Delete 事件持续写入 Elasticsearch。通过稳定的主键和文档 ID，可以构建面向检索、分析和在线查询的实时索引。

版本兼容性 #

特色功能 #

• 支持将 CDC 全量和增量数据持续写入 Elasticsearch，并通过稳定文档 ID 处理更新和删除。
• 支持通过 Pipeline route 将一张或多张源表映射到指定索引。
• 支持 Elasticsearch 6.x、7.x 和 8.x，并针对 Elasticsearch 8.x 提供并发请求、缓冲请求、批次字节数和单记录字节数参数。
• 支持按记录字段生成索引分片后缀，适用于仅同步 Insert/Update 且分片字段保持不变的场景。

前提条件 #

1. 已开通火山引擎流式计算 Flink 版，并创建 Flink 1.16 或 1.20 的项目和资源池；Flink CDC 版本选择 3.4.0 或以上。
2. 已准备 Elasticsearch 6.x、7.x 或 8.x 集群，并确保 Flink 资源池能够访问其地址和端口。
3. 已提前创建目标索引并规划 Mapping。Connector 与 Apache Flink CDC 3.4 社区行为一致，不自动创建 Elasticsearch 索引。
4. 已确认源表主键可以稳定生成目标文档 ID，以支持 Update、Delete 和故障恢复后的幂等覆盖。
5. 已获取 Elasticsearch 用户名和密码，并在 Flink 中创建加密变量，避免在 YAML 中保存明文凭证。
6. 已根据业务数据规划日期、Decimal、嵌套对象和高基数字段的 Mapping。
7. Elasticsearch 使用自签 CA 的 HTTPS 证书时，已具备下载 CA 证书并使用 keytool 生成 Java TrustStore 的环境。

使用限制 #

• Connector 不自动创建 Elasticsearch 索引；目标索引和 Mapping 必须提前准备。
• Connector 不向 Elasticsearch 应用 Schema 变更。新增、删除、重命名字段或修改字段类型时，只更新内部序列化 Schema，不会自动修改目标 Mapping。
• Elasticsearch Sink 不提供 Exactly-Once。启用 Checkpoint 后，应通过稳定文档 ID 使重放结果收敛。
• Elasticsearch 已存在字段的类型不能原地修改。不兼容类型变更应使用新索引，并通过 Reindex 或切换别名完成迁移。
• Update 和 Delete 依赖稳定主键。改变主键或 Route 规则可能产生新文档，或导致旧文档无法删除。
• Elasticsearch 8.x 的单条记录不得超过 record.size.max.bytes；批量大小同时受记录数、总字节数和缓冲时间控制。
• 在社区版 3.4 中，不同 Elasticsearch 版本支持的批量写入相关配置参数存在差异：
  • Elasticsearch 8.x：支持 inflight.requests.max、buffered.requests.max、batch.size.max.bytes、record.size.max.bytes。
  • Elasticsearch 6.x / 7.x：仅支持 batch.size.max、buffer.time.max.ms。
• MAP 和 ROW 会序列化为 JSON 字符串，目标 Mapping 必须与实际写入值匹配。
• 需要同步 Delete 的作业不能配置 sharding.suffix.key。社区 3.4 的 Delete 请求仍写向未添加后缀的索引，无法正确删除后缀索引中的文档。
• 仅同步 Insert/Update 且使用 sharding.suffix.key 时，分片字段在文档生命周期内必须保持不变。字段值改变会在新的后缀索引写入文档，旧索引中的文档不会自动删除。
• OpenSearch 尚未进行兼容性验证，本文不承诺支持。

语法结构 #

source: 
  type: <source_type>
  # Source 参数

sink:
  type: elasticsearch
  hosts: https://<elasticsearch_endpoint>:9200
  version: 8
  username: ${secret_values.es-username}
  password: ${secret_values.es-password}

route:
  - source-table: <source_database>.<source_table>
    sink-table: <target_index>

pipeline:
  name: <pipeline_name>
  parallelism: 2

参数说明 #

Sink 参数 #

类型映射 #

Elasticsearch Sink 先将 Flink CDC 数据序列化为 JSON，再写入目标索引。下表描述的是 CDC 类型到 JSON 值的映射，不等同于 Elasticsearch Mapping 类型。

使用示例 #

MySQL 同步到 Elasticsearch #

以下示例将 MySQL 表 <source_database>.<source_table> 的全量和增量数据写入 <target_index>：

source:
  type: mysql
  hostname: <mysql_host>
  port: 3306
  username: ${secret_values.mysql-username}
  password: ${secret_values.mysql-password} 
  tables: <source_database>.<source_table>
  server-id: 5400-5404

sink:
  type: elasticsearch
  name: Elasticsearch Sink
  hosts: https://<elasticsearch_endpoint>:9200
  version: 8
  username: ${secret_values.es-username}
  password: ${secret_values.es-password}
  batch.size.max: 500
  batch.size.max.bytes: 5242880
  buffer.time.max.ms: 5000
  record.size.max.bytes: 10485760

route:
  - source-table: <source_database>.<source_table>
    sink-table: <target_index>
    description: Route source table to Elasticsearch index

pipeline:
  name: <pipeline_name>
  parallelism: 2

version 必须与 Elasticsearch 集群大版本一致。未配置 route 时，默认索引名为源表完整 TableId，即 namespace.schemaName.tableName；生产环境建议显式配置目标索引。
预期结果： 全量阶段将源表已有数据写入目标索引；增量阶段持续处理新增、更新和删除事件。同一主键的更新覆盖原文档，删除事件删除相同文档 ID 的文档。

配置 HTTPS 自签 CA #

以下配置方式尚未在当前产品环境完成验证。正式使用前，请先在测试作业中验证 TrustStore 文件的分发路径和 JVM 参数。

1. 在 Elasticsearch 控制台下载 CA 证书 ca.cer。
2. 将 CA 证书导入并生成 JKS 格式的 TrustStore：

keytool -importcert \
  -alias es-ca \ 
  -file ca.cer \
  -keystore es-truststore.jks \
  -storepass changeit \
  -noprompt

1. 将 es-truststore.jks 放入 Flink 作业依赖目录，与 Connector JAR 同级，并确保该文件随作业一起分发。
2. 在作业“自定义参数”中追加以下配置：

env.java.opts: -Djavax.net.ssl.trustStore=/opt/tiger/workdir/es-truststore.jks -Djavax.net.ssl.trustStorePassword=changeit 

生产环境应将示例密码 changeit 替换为自定义强密码，并同步修改生成 TrustStore 和 env.java.opts 时使用的密码。修改后重启作业，使 JVM 参数生效。

运行机制 #

Source 读取存量数据和增量变更，Transform 完成字段投影、过滤或类型调整，Route 将源表映射到目标索引，Elasticsearch Sink 将行变更序列化为 JSON 文档并执行写入或删除。
Insert 和 Update 使用源表主键生成稳定文档 ID，故障恢复期间的重复写入可以覆盖同一文档。Delete 使用相同文档 ID 删除目标文档。Connector 采用 At-Least-Once 语义，正确处理更新、删除和恢复的前提是主键及 Route 规则保持稳定。
Schema 事件只更新 Connector 的内部序列化 Schema，不修改 Elasticsearch Mapping。新增字段能否被目标索引接受取决于现有 Mapping 和动态 Mapping 配置；生产环境应先调整索引模板或 Mapping，再放行 Schema 变更。

验证与排障 #

验证方法 #

1. 启动作业并等待全量阶段完成，核对源表行数与目标索引文档数。
2. 新增一条源记录，确认目标索引出现对应文档。
3. 更新同一主键，确认原文档被覆盖，而不是新增另一条文档。
4. 删除同一主键，确认对应目标文档被删除。
5. 等待 Checkpoint 成功后执行受控重启，确认恢复后同一主键仍对应单一文档。
6. 检查日期、Decimal、keyword、text 和嵌套字段的实际 Mapping，并确认日志中没有持续出现 mapper_parsing_exception。

常见问题 #

注意事项 #

• 不要在需要同步 Delete 的作业中使用 sharding.suffix.key。
• 目标索引 Mapping 应在生产写入前完成规划和验证，避免由动态 Mapping 错误推断字段类型。
• 用户名、密码和其他凭证必须通过 Flink 加密变量引用，不要明文写入 YAML。
• 自签 CA TrustStore 配置尚未在当前产品环境完成验证，正式使用前必须先在测试作业验证。

相关文档 #

• Apache Flink CDC 3.4：Elasticsearch Pipeline Connector
• Apache Flink CDC 3.4：Pipeline Connector 版本矩阵
• Apache Flink CDC 3.4：ElasticsearchDataSinkOptions API
• 火山引擎流式计算 Flink 版：使用变量
• 火山引擎流式计算 Flink 版：计算任务基础操作