一、前置准备
·已部署可用的 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-5404sink: type: elasticsearch name: Elasticsearch Sink hosts: http://127.0.0.1:9092,http://127.0.0.1:9093 version: 8pipeline: 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 | – |
| MAPROW | 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 生效。