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 表：

1. CREATE TABLE myUserTable (
2.   user_id STRING,
3.   user_name STRING,
4.   uv BIGINT,
5.   pv BIGINT,
6.   PRIMARY KEY (user_id) NOT ENFORCED
7. ) WITH (
8.   'connector' = 'elasticsearch-7',
9.   'hosts' = 'http://localhost:9200',
10.   'index' = 'users'
11. );

复制代码

三、连接器参数

参数是否必选默认值数据类型描述connector必选(none)String指定要使用的连接器，有效值为：elasticsearch-6：连接到 Elasticsearch 6.x 的集群。elasticsearch-7：连接到 Elasticsearch 7.x 及更高版本的集群。hosts必选(none)String要连接到的一台或多台 Elasticsearch 主机，比方 ‘http://host_name:9092;http://host_name:9093’。index必选(none)StringElasticsearch 中每条记录的索引。可以是一个静态索引（比方 ‘myIndex’）或一个动态索引（比方 'index-{log_tsdocument-type6.x 版本中必选(none)StringElasticsearch 文档类型。在 elasticsearch-7 中不再必要。document-id.key-delimiter可选-String复合键的分隔符（默以为"_“），比方，指定为”                                                  "                                  将导致文档                                  I                                  D                                  为                                  "                                  K                                  E                                  Y                                  1                                          "将导致文档 ID 为"KEY1                           "将导致文档ID为"KEY1KEY2$KEY3"。username可选(none)String用于连接 Elasticsearch 实例的用户名。请留意，Elasticsearch 没有预绑定安全特性，但你可以通过如下指南启用它来保护 Elasticsearch 集群。password可选(none)String用于连接 Elasticsearch 实例的密码。如果设置了username，则此选项也必须设置为非空字符串。failure-handler可选failString对 Elasticsearch 哀求失败情况下的失败处置惩罚策略。有效策略为：fail：如果哀求失败并因此导致作业失败，则抛出异常。ignore：忽略失败并放弃哀求。retry-rejected：重新添加由于队列容量饱和而失败的哀求。自定义类名称：使用 ActionRequestFailureHandler 的子类进行失败处置惩罚。sink.delivery-guarantee可选AT_LEAST_ONCEString承诺时可选择交付包管。有效值为：EXACTLY_ONCE：在故障转移情况下，记录也仅传递一次。AT_LEAST_ONCE：确保记录被传递，但大概会发生同一条记录被传递多次的情况。NONE：努力提供记录。sink.flush-on-checkpoint可选trueBoolean在进行 checkpoint 时是否包管刷出缓冲区中的数据。如果关闭这一选项，在进行checkpoint时 sink 将不再为所有进行 中的哀求等待 Elasticsearch 的实行完成确认。因此，在这种情况下 sink 将不对至少一次的哀求的同等性提供任何包管。sink.bulk-flush.max-actions可选1000Integer每个批量哀求的最大缓冲操纵数。 可以设置为’0’来禁用它。sink.bulk-flush.max-size可选2mbMemorySize每个批量哀求的缓冲操纵在内存中的最大值。单元必须为 MB。 可以设置为’0’来禁用它。sink.bulk-flush.interval可选1sDurationflush 缓冲操纵的隔断。 可以设置为’0’来禁用它。留意，'sink.bulk-flush.max-size’和’sink.bulk-flush.max-actions’都设置为’0’的这种 flush 隔断设置答应对缓冲操纵进行完全异步处置惩罚。sink.bulk-flush.backoff.strategy可选DISABLEDString指定在由于暂时哀求错误导致任何 flush 操纵失败时如何实行重试。有效策略为：DISABLED：不实行重试，即第一次哀求错误后失败。CONSTANT：等待重试之间的回退延迟。EXPONENTIAL：先等待回退延迟，然后在重试之间指数递增。sink.bulk-flush.backoff.max-retries可选(none)Integer最大回退重试次数。sink.bulk-flush.backoff.delay可选(none)Duration每次退避实验之间的延迟。对于 CONSTANT 退避策略，该值是每次重试之间的延迟。对于 EXPONENTIAL 退避策略，该值是初始的延迟。connection.path-prefix可选(none)String添加到每个 REST 通信中的前缀字符串，比方，‘/v1’。connection.request-timeout可选(none)Duration从连接管理器哀求连接的超时时间。超时时间必须大于大概即是 0，如果设置为 0 则是无穷超时。connection.timeout可选(none)Duration创建哀求的超时时间 。超时时间必须大于大概即是 0 ，如果设置为 0 则是无穷超时。socket.timeout可选(none)Duration等待数据的 socket 的超时时间 (SO_TIMEOUT)。超时时间必须大于大概即是 0，如果设置为 0 则是无穷超时。format可选jsonStringElasticsearch 连接器支持指定格式。该格式必须天生一个有效的 json 文档。 默认使用内置的 ‘json’ 格式。 四、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 中的数据类型的映射关系。
Flink SQL类型JSON类型CHAR/VARCHAR/STRINGstringBOOLEANbooleanBINARY/VARBINARYstring with encoding: base64DECIMALnumberTINYINTnumberSMALLINTnumberINTnumberBIGINTnumberFLOATnumberDOUBLEnumberDATEstring with format: dateTIMEstring with format: timeTIMESTAMPstring with format: date-timeTIMESTAMP_WITH_LOCAL_TIME_ZONEstring with format: date-time (with UTC time zone)INTERVALnumberARRAYarrayMAP / MULTISETobjectROWobject
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。