Flink系列之：Elasticsearch SQL 连接器-灵析社区

一、Elasticsearch SQL 连接器

Sink: Batch
Sink: Streaming Append & Upsert Mode
Elasticsearch 连接器允许将数据写入到 Elasticsearch 引擎的索引中。本文档描述运行 SQL 查询时如何设置 Elasticsearch 连接器。
连接器可以工作在 upsert 模式，使用 DDL 中定义的主键与外部系统交换 UPDATE/DELETE 消息。
如果 DDL 中没有定义主键，那么连接器只能工作在 append 模式，只能与外部系统交换 INSERT 消息。

二、创建 Elasticsearch表

以下示例展示了如何创建 Elasticsearch sink 表：

CREATE TABLE myUserTable (
  user_id STRING,
  user_name STRING,
  uv BIGINT,
  pv BIGINT,
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'connector' = 'elasticsearch-7',
  'hosts' = 'http://localhost:9200',
  'index' = 'users'
);

三、连接器参数

四、Key 处理

Elasticsearch sink 可以根据是否定义了一个主键来确定是在 upsert 模式还是 append 模式下工作。如果定义了主键，Elasticsearch sink 将以 upsert 模式工作，该模式可以消费包含 UPDATE/DELETE 消息的查询。如果未定义主键，Elasticsearch sink 将以 append 模式工作，该模式只能消费包含 INSERT 消息的查询。
在 Elasticsearch 连接器中，主键用于计算 Elasticsearch 的文档 id，文档 id 为最多 512 字节且不包含空格的字符串。 Elasticsearch 连接器通过使用 document-id.key-delimiter 指定的键分隔符按照 DDL 中定义的顺序连接所有主键字段，为每一行记录生成一个文档 ID 字符串。某些类型不允许作为主键字段，因为它们没有对应的字符串表示形式，例如，BYTES，ROW，ARRAY，MAP 等。如果未指定主键，Elasticsearch 将自动生成文档 id。

五、动态索引

Elasticsearch sink 同时支持静态索引和动态索引。
如果你想使用静态索引，则 index 选项值应为纯字符串，例如 ‘myusers’，所有记录都将被写入到 “myusers” 索引中。
如果你想使用动态索引，你可以使用 {field_name} 来引用记录中的字段值来动态生成目标索引。你也可以使用 ‘{field_name|date_format_string}’ 将 TIMESTAMP/DATE/TIME 类型的字段值转换为 date_format_string 指定的格式。 date_format_string 与 Java 的 DateTimeFormatter 兼容。例如，如果选项值设置为 ‘myusers-{log_ts|yyyy-MM-dd}’，则 log_ts 字段值为 2020-03-27 12:25:55 的记录将被写入到 “myusers-2020-03-27” 索引中。
你也可以使用 ‘{now()|date_format_string}’ 将当前的系统时间转换为 date_format_string 指定的格式。now() 对应的时间类型是 TIMESTAMP_WITH_LTZ 。在将系统时间格式化为字符串时会使用 session 中通过 table.local-time-zone 中配置的时区。使用 NOW(), now(), CURRENT_TIMESTAMP, current_timestamp 均可以。
注意: 使用当前系统时间生成的动态索引时，对于 changelog 的流，无法保证同一主键对应的记录能产生相同的索引名, 因此使用基于系统时间的动态索引，只能支持 append only 的流。

六、数据类型映射

Elasticsearch 将文档存储在 JSON 字符串中。因此数据类型映射介于 Flink 数据类型和 JSON 数据类型之间。 Flink 为 Elasticsearch 连接器使用内置的 ‘json’ 格式。

下表列出了 Flink 中的数据类型与 JSON 中的数据类型的映射关系。