本文为您介绍Elasticsearch(ES)结果表DDL定义、WITH参数、类型映射和代码示例。

说明 该Connector支持的功能是和阿里云Elasticsearch产品对齐的,暂不支持HTTPS加密访问形式。

DDL定义

 CREATE TABLE es_sink(
   user_id   STRING,
   user_name   STRING,
   uv BIGINT,
   pv BIGINT,
   PRIMARY KEY (user_id) NOT ENFORCED  -- 主键可选,如果定义了主键,则作为文档ID,否则文档ID将为随机值。
) WIHT (
   'connector' = 'elasticsearch-7',
   'hosts' = '<yourHosts>',
   'index' = '<yourIndex>',
   'document-type' = '<yourEelasticsearch.types>',
   'username' ='<yourElasticsearch.accessId>',
   'password' ='<yourElasticsearch.accessKey>'
);
说明
  • 仅支持Elasticsearch 6.0和7.0版本。
  • DDL中的字段均对应Elasticsearch文档中的字段,不支持将文档ID写入表中。

WITH参数

参数 说明 是否必填 备注
connector 结果表类型 固定值为elasticsearch-6elasticsearch-7
hosts Server地址 例如:127.0.0.1:9200
index 文档索引名称 默认为空,不进行权限验证。
document-type Type名称
  • elasticsearch-6:必填
  • elasticsearch-7:不支持
username 访问实例ID 默认为空,不进行权限验证。
password 访问实例密钥 如果定义了username,则必须定义非空的password。
document-id.key-delimiter 文档ID的分隔符 默认值为_
failure-handler Elasticsearch请求失败时的故障处理策略。 可选策略如下:
  • fail(默认值):如果请求失败,则作业失败。
  • igonre:忽略失败并删除请求。
  • retry_rejected:重新添加由于队列容量满而失败的请求。
  • custom class name:用于使用ActionRequestFailureHandler子类进行故障处理。
sink.flush-on-checkpoint 是否在checkpoint时执行flush。 默认值为true。禁用该功能后,在Elasticsearch进行Checkpoint时,connector将不等待确认所有pending请求已完成。因此,connector不会为请求提供at-least-once保证。
sink.bulk-flush.backoff.strategy 如果由于临时请求错误导致flush操作失败,则设置sink.bulk-flush.backoff.strategy指定重试策略。
  • DISABLED(默认值):不执行重试,即第一次请求错误后失败。
  • CONSTANT:常量回退。
  • EXPONENTIAL:指数回退。
sink.bulk-flush.backoff.max-retries 最大回退重试次数。 默认值为8。
sink.bulk-flush.backoff.delay 每次回退尝试之间的延迟。
  • CONSTANT:每次重试之间的延迟。
  • EXPONENTIAL:这是初始基准延迟。
sink.bulk-flush.max-actions 每个批量请求的最大缓冲操作数。 默认值为1000,0表示禁用该功能。
sink.bulk-flush.max-size 存放请求的缓冲区内存最大值。 单位为MB,默认值为2,0表示禁用该功能。
sink.bulk-flush.interval flush的间隔。 默认值为1s,0表示禁用该功能。
connection.path-prefix 要添加到每个REST通信中的前缀字符串。 默认值为空。

文档ID

Elasticsearch Sink可以根据是否定义了主键确定是其在upsert模式还是在append模式下工作。如果定义了主键,Elasticsearch Sink将在upsert模式下工作,该模式可以消费包含UPDATE和DELETE的消息。如果未定义主键,Elasticsearch Sink将以append模式工作,该模式只能消费INSERT消息。

在Elasticsearch Sink中,主键用于计算Elasticsearch的文档ID。文档ID为最多512个字节不包含空格的字符串。Elasticsearch Sink通过使用document-id.key-delimiter指定的键分隔符按照DDL中定义的顺序连接所有主键字段,从而为每一行生成一个文档ID字符串。某些类型(例如BYTES、ROW、ARRAY和MAP等)由于没有对应的字符串表示形式,所以不允许其作为主键字段。如果未指定主键,Elasticsearch将自动生成随机的文档ID。

动态索引

Elasticsearch Sink同时支持静态索引和动态索引:
  • 如果使用静态索引,则索引选项值应为纯字符串,例如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索引。

类型映射

Flink全托管以JSON来解析Elasticsearch数据,详情请参见数据类型映射关系

代码示例

CREATE TEMPORARY TABLE datagen_source (
  id STRING, 
  name STRING,
  uv BIGINT
) with (
  'connector' = 'datagen'
);

CREATE TEMPORARY TABLE es_sink (
   user_id STRING,
   user_name STRING,
   uv BIGINT,
   PRIMARY KEY (user_id) NOT ENFORCED -- 主键可选,如果定义了主键,则作为文档ID,否则文档ID将为随机值。
) WITH (
   'connector' = 'elasticsearch-7',
   'hosts' = '<yourHosts>',
   'index' = '<yourIndex>',
   'document-type' = '<yourElasticsearch.types>',
   'username' ='<yourElasticsearch.accessId>',
   'password' ='<yourElasticsearch.accessKey>'
);

INSERT INTO es_sink
   SELECT id, name, uv
FROM datagen_source;