Elasticsearch SQL 连接器

FLINK-1.14

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',
  'version' = '7'
  'hosts' = 'http://localhost:9200',
  'index' = 'users'
);

带认证

CREATE TABLE es_sink_table (
  testInt INT,
  testVarchar VARCHAR
) WITH (
    'connector' = 'elasticsearch',
    'version' = '7',
    'hosts' = 'http://localhost:29200',
    'index' = 'users',
    'format' = 'json',
    'username' = 'elastic',
    'password' = 'xxx'
);

参数配置

特性

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。

有关 PRIMARY KEY 语法的更多详细信息，请参见 [CREATE TABLE DDL]($flink-docs-connectors-{{< ref "docs/dev/table/sql/create" >}}#create-table)。

动态索引

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" 索引中。

数据类型映射

Elasticsearch 将文档存储在 JSON 字符串中。因此数据类型映射介于 Flink 数据类型和 JSON 数据类型之间。 Flink 为 Elasticsearch 连接器使用内置的 'json' 格式。更多类型映射的详细信息，请参阅JSON Format

Flink-1.12

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',
  'version' = '7'
  'hosts' = 'http://localhost:9200',
  'index' = 'users'
);

带认证

CREATE TABLE es_sink_table (
  testInt INT,
  testVarchar VARCHAR
) WITH (
    'connector' = 'elasticsearch',
    'version' = '7',
    'hosts' = 'http://localhost:29200',
    'index' = 'users',
    'format' = 'json',
    'username' = 'elastic',
    'password' = 'xxx'
);

维表

支持将es作为维表使用

create table t1 (testInt int, c as testInt%2, proc as proctime()) with ('connector'='datagen');

CREATE TABLE dim (
  testInt INT,
  testVarchar varchar
) WITH (
    'connector' = 'elasticsearch',
    'version' = '7',
    'hosts' = 'http://ip:29200',
    'index' = 'users',
    'document-type' = '_doc',
    'format' = 'json',
    'username' = 'elastic',
    'password' = '加密内容1',
    'lookup.cache.max-rows' = '100',
    'lookup.cache.ttl' = '1200s',
    'lookup.max-retries' = '3'
);

create table t2
(testInt int,
testVarchar varchar)
with ('connector'='print');

INSERT INTO t2
SELECT
    t1.c as testInt,
    d.testVarchar
FROM t1 left join dim for system_time as of t1.proc as d on t1.c = d.testInt;

特性

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。

有关 PRIMARY KEY 语法的更多详细信息，请参见 [CREATE TABLE DDL]($flink-docs-connectors-{{< ref "docs/dev/table/sql/create" >}}#create-table)。

动态索引

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" 索引中。

数据类型映射

Elasticsearch 将文档存储在 JSON 字符串中。因此数据类型映射介于 Flink 数据类型和 JSON 数据类型之间。 Flink 为 Elasticsearch 连接器使用内置的 'json' 格式。更多类型映射的详细信息，请参阅JSON Format

Flink-1.10

ES 5 Sink

简述

Easystream 支持输出到 ES 5。

示例

CREATE TABLE MyUserTable (
  ...
) WITH (
  'connector.type' = 'elasticsearch',
  'connector.version' = '5',        
  'connector.hosts' = 'http://host_name:9092;http://host_name:9093',
  'connector.cluster.name' = 'escluster',
  'connector.index' = 'MyUsers',     
  'connector.document-type' = 'user', 
  'update-mode' = 'append',                    
  'connector.key-delimiter' = '$',  
  'connector.key-null-literal' = 'n/a',
  'connector.failure-handler' = '...',  
  'connector.flush-on-checkpoint' = 'true',                                                          
  'connector.bulk-flush.max-actions' = '42', 
  'connector.bulk-flush.max-size' = '42 mb', 
  'connector.bulk-flush.interval' = '60000',
  'connector.bulk-flush.back-off.type' = '...',      
  'connector.bulk-flush.back-off.max-retries' = '3',
  'connector.bulk-flush.back-off.delay' = '30000',
  'connector.connection-max-retry-timeout' = '3',   
  'connector.connection-path-prefix' = '/',
  'format.type' = 'json',
  'format.derive-schema' = 'true'
  )

• Flink Elasticsearch Connector 官方文档

With 参数

ES 6/7 Sink

简述

Easystream 支持输出到 ES 6 或 7。

示例

CREATE TABLE MyUserTable (
  ...
) WITH (
  'connector.type' = 'elasticsearch',
  'connector.version' = '6',        
  'connector.hosts.0.hostname' = 'host_name',
  'connector.hosts.0.protocol' = 'http',
  'connector.index' = 'MyUsers',     
  'connector.document-type' = 'user', 
  'update-mode' = 'append',                    
  'connector.key-delimiter' = '$',  
  'connector.key-null-literal' = 'n/a',
  'connector.failure-handler' = '...',  
  'connector.flush-on-checkpoint' = 'true',                                                          
  'connector.bulk-flush.max-actions' = '42', 
  'connector.bulk-flush.max-size' = '42 mb', 
  'connector.bulk-flush.interval' = '60000',
  'connector.bulk-flush.back-off.type' = '...',      
  'connector.bulk-flush.back-off.max-retries' = '3',
  'connector.bulk-flush.back-off.delay' = '30000',
  'connector.connection-max-retry-timeout' = '3',   
  'connector.connection-path-prefix' = '/',
  'format.type' = 'json',
  'format.derive-schema' = 'true'
  )

• Flink Elasticsearch Connector 官方文档

With 参数