帮你快速理解、总结文档立即下载
文档中心>流计算 Oceanus>CDC 数据同步开发指南>上下游开发指南>Elasticsearch 连接器

Elasticsearch 连接器

最近更新时间：2026-10-15 17:58:30
我的收藏

本页目录：

Elasticsearch 连接器可以作为 Sink（type: elasticsearch），把上游的变更事件写入 Elasticsearch，提供数据实时同步能力。


前置准备

已部署可用的 Elasticsearch 集群，版本为 6.x、7.x 或 8.x。
作业运行环境需能访问 Elasticsearch 的 HTTP 端口，并具备目标索引的写入权限。
目标索引需要提前创建。连接器不会自动创建索引，也不会同步表结构变更到 Elasticsearch。


版本说明

Flink 版本
说明
1.20
基于 Flink-cdc 社区 release-3.6 开发，JDK 版本为 11，支持 Elasticsearch 版本为 6.x、7.x、8.x

使用范围

可作为 Sink（type: elasticsearch）使用。


快速示例

从 MySQL 同步到 Elasticsearch：
source:
  type: mysql
  name: MySQL Source
  hostname: 127.0.0.1
  port: 3306
  username: admin
  password: pass
  tables: adb.\\.*
  server-id: 5401-5404

sink:
  type: elasticsearch
  name: Elasticsearch Sink
  hosts: http://127.0.0.1:9092,http://127.0.0.1:9093
  version: 8

pipeline:
  name: MySQL to Elasticsearch Pipeline
  parallelism: 2

通过 route 把多张表写入同一个索引：
route:
  - source-table: adb.\\.*
    sink-table: default_index
    description: sync adb.\\.* tables to default_index


Sink 配置项

参数
是否必填
默认值
类型
说明
hosts
required
(none)
String
Elasticsearch 的主机地址列表，多个之间用英文逗号分隔。
例如 http://127.0.0.1:9092,http://127.0.0.1:9093。
version
required
7
Integer
Elasticsearch 的版本，可取 6、7 或 8。
需要显式配置。
username
optional
(none)
String
认证用户名。未配置时不启用认证。
password
optional
(none)
String
认证密码。
batch.size.max
optional
500
Integer
单个批量请求缓冲的最大操作数。
inflight.requests.max
optional
5
Integer
允许并发执行的最大请求数。
buffered.requests.max
optional
1000
Integer
内存中缓冲的最大请求数。
batch.size.max.bytes
optional
10485760（10MB）
Long
单个批量请求的最大字节数。
buffer.time.max.ms
optional
5000
Long
未填满的批次在被强制发送前的最大等待时间，单位为毫秒。
record.size.max.bytes
optional
10485760（10MB）
Long
单条记录的最大字节数。
ssl.verify
optional
true
Boolean
通过 HTTPS 连接时是否校验证书。设为 false 时信任所有证书，并跳过主机名校验， 不建议在生产环境使用。
设为 true 但未配置 ssl.cert 时，使用运行环境的默认信任库校验。
ssl.cert
optional
(none)
String
用于校验 HTTPS 服务端证书的 CA 证书（PEM / X.509 格式）。
具体含义由 ssl.cert.type 决定：FILE 时为文件路径， BASE64 时为 Base64 编码的证书内容。
仅在 ssl.verify 为 true 时生效。
ssl.cert.type
optional
FILE
String
ssl.cert 的解读方式，可取 FILE（每个 TaskManager 节点上的本地文件路径） 或 BASE64（内联的 Base64 证书内容）。
取值大小写不敏感。
ssl.hostname.verification
optional
false
Boolean
是否校验证书中的主机名与实际连接的主机一致。默认关闭， 以支持证书 CN 为主机名、但按 IP 连接的场景。
仅在 ssl.verify 为 true 且配置了 ssl.cert 时生效。
sharding.suffix.key
optional
(空)
String
为表指定分索引后缀所依据的字段，详见「分索引」章节。
sharding.suffix.separator
optional
_
String
表名与分索引后缀之间的分隔符。必须为小写， 且不能包含 \\、/、*、?、 "、<、>、|、空格、 ,、# 这些字符。
Sink 侧的时区取自 pipeline 配置 local-time-zone，未配置时使用作业运行环境的系统时区。


分索引

通过 sharding.suffix.key 可以让同一张表的数据按某个字段的值写入不同的索引，索引名为「表名 + 分隔符 + 字段值」。多条配置之间用英文分号（;）分隔，表名与字段之间用英文冒号（:）分隔。
sink:
  type: elasticsearch
  sharding.suffix.key: table1:col1;table2:col2
  sharding.suffix.separator: '-'

配置时需要注意：

表名需要与上游表的标识一致，格式为 namespace.tableName。
格式不符合要求时作业启动失败。
未配置 sharding.suffix.key 时，如果上游表结构本身带有分区键，会使用第一个分区键的值作为后缀。
分隔符只能使用小写字母及 Elasticsearch 索引名允许的字符，否则作业启动失败。
分索引后缀只作用于写入和更新，删除操作使用的索引名不带后缀。开启分索引后，上游的删除操作会作用在不带后缀的索引上，需要留意这一点对数据的影响。


数据写入

连接器按下列规则把变更事件写入 Elasticsearch：

写入索引名：未配置分索引时为上游表名（经 route 转换后的目标表名）；配置分索引时为「表名 + 分隔符 + 字段值」，字段值取不到时使用表名。
文档 ID（_id）：由上游表主键的各字段值用下划线（_）连接而成。上游表没有主键时文档 ID 为空字符串，同一张表的所有记录会共用该 ID 而互相覆盖，因此请同步包含主键的源表。
INSERT、UPDATE、REPLACE 写入整行数据；DELETE 按文档 ID 删除对应文档。


表结构变更同步

Elasticsearch Sink 不向 Elasticsearch 下发任何表结构变更，也不会自动创建索引。上游的表结构变更仅用于连接器内部更新字段的序列化方式，以保证新增、删除、重命名字段后写入的数据与最新表结构一致。

因此需要注意：

目标索引需要提前创建。
上游新增字段后，Elasticsearch 侧如果是严格映射（dynamic 为 strict），需要在目标索引中同步添加对应字段，否则写入会失败。
某张表的建表事件尚未到达时，如果该表先收到数据变更或增删列、重命名列的事件，连接器会因找不到该表的表结构而报错。正常情况下建表事件先于其它事件到达，不会出现该问题。


数据类型映射

CDC 类型
JSON 类型
说明
CHAR(n)
VARCHAR(n)
STRING
string
-
BOOLEAN
boolean
-
BINARY(n)
VARBINARY(n)
BYTES
string
字节数组序列化为 Base64 编码的字符串。
DECIMAL(p, s)
string
按 decimal 的字符串形式写入，避免精度损失。
TINYINT
number
-
SMALLINT
number
-
INT
number
-
BIGINT
number
-
FLOAT
number
-
DOUBLE
number
-
DATE
string
格式为 yyyy-MM-dd。
TIMESTAMP(p)
string
格式为 yyyy-MM-dd HH:mm:ss.SSSSSS。
TIMESTAMP_LTZ(p)
string
先按 pipeline 的 local-time-zone 换算为本地时间， 格式为 yyyy-MM-dd HH:mm:ss.SSSSSS。
TIMESTAMP WITH TIME ZONE(p)
string
先按 pipeline 的 local-time-zone 换算， 格式为 yyyy-MM-dd HH:mm:ss.SSSSSS。
ARRAY
array
-
MAP
ROW
string
序列化为 JSON 字符串后写入。
上表列出的是写入 Elasticsearch 文档时各字段对应的 JSON 类型。 Elasticsearch 侧的实际字段类型由目标索引的 mapping 决定，连接器不写入类型信息,因此需要在 mapping 中为这些字段配置匹配的类型。

注意事项

hosts 与 version 为必填参数，未配置时作业启动失败。
目标索引需要提前创建，连接器不会自动创建索引。
上游表需要包含主键，连接器使用主键值拼接文档 ID。
时间类型统一以字符串形式写入，需要在目标索引的映射中把对应字段设为 date 类型并指定匹配的时间格式，否则会被当作普通字符串处理。
batch.size.max.bytes、inflight.requests.max、buffered.requests.max、record.size.max.bytes 仅在 version 为 8 时生效；version 为 6 或 7 时，只有 batch.size.max 与 buffer.time.max.ms 生效。
中国站