流计算 Oceanus 数据分析引擎 Elasticsearch Service

二、版本说明

注意：Flink 1.18 及以上版本中，Elasticsearch Connector 基于新的 SinkV2 API 实现。ES6/ES7 的参数与旧版本兼容，ES8 使用独立的 connector 标识 elasticsearch-8，参数体系有较大变化，详见下方各节说明。

三、使用范围

Elasticsearch 支持写入，可以作为 Tuple 数据流的目的表（Sink），也可以作为 Upsert 数据流的目的表（Sink，自动以文档 _id 字段生成主键，并更新之前的文档版本）。

如果希望将 JDBC 数据库的变动记录，将其作为流式源表消费，可以使用 Debezium、Canal 等，对 JDBC 数据库的变更进行捕获和订阅，然后 Flink 即可对这些变更事件进行进一步的处理。可参见 Kafka。

Flink 1.13 支持 Elasticsearch 的批模式读，仅支持 Elasticsearch 7。

Flink 1.18 及以上版本支持将 Elasticsearch 作为维表（Lookup Table），用于流表 JOIN 关联查询，支持 ES 6.x、7.x 和 8.x 版本。

五、用作 Elasticsearch 6 数据目的（Sink）

CREATE TABLE elasticsearch6_sink_table ( `id` INT, `name` STRING, PRIMARY KEY (`id`) NOT ENFORCED — 对应 Elasticsearch 中的 _id) WITH ( 'connector' = 'elasticsearch-6', — 输出到 Elasticsearch 6 'username' = '$username', — 选填 用户名 'password' = '$password', — 选填 密码 'hosts' = 'http://xx.xx.xx.xx:9200', — Elasticsearch 的连接地址 'index' = 'my-index', — Elasticsearch 的 Index 名 'document-type' = '_doc', — Elasticsearch 的 Document 类型 'format' = 'json' — 输出数据格式，目前只支持 'json');

六、用作 Elasticsearch 7 数据目的（Sink）

CREATE TABLE elasticsearch7_sink_table ( `id` INT, `name` STRING, PRIMARY KEY (`id`) NOT ENFORCED — 对应 Elasticsearch 中的 _id) WITH ( 'connector' = 'elasticsearch-7', — 输出到 Elasticsearch 7 'username' = '$username', — 选填 用户名 'password' = '$password', — 选填 密码 'hosts' = 'http://xx.xx.xx.xx:9200', — Elasticsearch 的连接地址 'index' = 'my-index', — Elasticsearch 的 Index 名 'format' = 'json' — 输出数据格式，目前只支持 'json');

七、作为 Elasticsearch 7 批数据源（Source）

CREATE TABLE elasticsearch7_source_table ( `id` bigint, `event_date` int, `app` int, primary key (`id`) not enforced) with ( — 必填参数 'connector' = 'es-source', 'endPoint' = 'xx.xx.xx.xx', — Elasticsearch 的连接 ip 'accessId' = 'elastic', — 用户名 'accessKey' = 'PASSWORD', — 密码 'indexName' = 'my-index', — Elasticsearch 的 Index 名 'format' = 'json', — 数据格式，只支持 'json' — 可选参数 'scheme' = 'http', — 连接协议 'port' = '9200', — 端口 'batchSize' = '2000', — 每个 scroll 请求从 Elasticsearch 集群获取的最大文档数 'keepScrollAliveSecs' = '60' — scroll 上下文保留的最长时间，单位为分钟);

八、用作 Elasticsearch 8 数据目的（Sink）

注意：说明：

注意：适用于 Flink 1.18 及以上版本。

CREATE TABLE elasticsearch8_sink_table ( `id` INT, `name` STRING, PRIMARY KEY (`id`) NOT ENFORCED — 对应 Elasticsearch 中的 _id) WITH ( 'connector' = 'elasticsearch-8', — 输出到 Elasticsearch 8 'username' = '$username', — 选填 用户名 'password' = '$password', — 选填 密码 'hosts' = 'http://xx.xx.xx.xx:9200', — Elasticsearch 的连接地址 'index' = 'my-index', — Elasticsearch 的 Index 名 'format' = 'json' — 输出数据格式，目前只支持 'json');

九、用作 Elasticsearch 6/7 维表（Lookup Table）

注意：说明：

注意：适用于 Flink 1.18 及以上版本。支持通过维表 JOIN 方式关联查询 Elasticsearch 中的数据进行字段补全。

CREATE TABLE elasticsearch7_lookup_table ( `id` INT, `name` STRING, PRIMARY KEY (`id`) NOT ENFORCED) WITH ( 'connector' = 'elasticsearch-7', 'username' = '$username', — 选填 用户名 'password' = '$password', — 选填 密码 'hosts' = 'http://host1:9200', 'index' = 'dim_user', 'lookup.cache.type' = 'PARTIAL', 'lookup.partial-cache.max-rows' = '200000', 'lookup.partial-cache.expire-after-write' = '30 min', 'lookup.partial-cache.expire-after-access' = '30 min', 'lookup.partial-cache.cache-missing-key' = 'true', 'lookup.max-retries' = '3' );

十一、作为数据目的（Elasticsearch 6/7）

注意：说明：

注意：此参数表同时适用于 Flink 1.16 及之前版本和 Flink 1.18+ 版本（ES6/ES7 的 elasticsearch-6 / elasticsearch-7 connector），两个版本参数基本一致。

十二、作为数据目的（Elasticsearch 8）

注意：说明：

注意：适用于 Flink 1.18 及以上版本，connector 标识为 elasticsearch-8。

十三、作为数据源

十四、作为维表

十六、作为数据目的

CREATE TABLE datagen_source_table ( id INT, name STRING ) WITH ( 'connector' = 'datagen', 'rows-per-second'='1' — 每秒产生的数据条数);CREATE TABLE elasticsearch7_sink_table ( `id` INT, `name` STRING) WITH ( 'connector' = 'elasticsearch-7', — 输出到 Elasticsearch 7 'username' = '$username', — 选填 用户名 'password' = '$password', — 选填 密码 'hosts' = 'http://xx.xx.xx.xx:9200', — Elasticsearch 的连接地址 'index' = 'my-index', — Elasticsearch 的 Index 名 'sink.bulk-flush.max-actions' = '1000', — 数据刷新频率 'sink.bulk-flush.interval' = '1s', — 数据刷新周期 'format' = 'json' — 输出数据格式，目前只支持 'json');INSERT INTO elasticsearch7_sink_table select * from datagen_source_table;

十七、作为数据目的（Elasticsearch 8）

注意：说明：

注意：适用于 Flink 1.18 及以上版本。

CREATE TABLE datagen_source_table ( id INT, name STRING ) WITH ( 'connector' = 'datagen', 'rows-per-second'='1' — 每秒产生的数据条数);CREATE TABLE elasticsearch8_sink_table ( `id` INT, `name` STRING, PRIMARY KEY (`id`) NOT ENFORCED) WITH ( 'connector' = 'elasticsearch-8', — 输出到 Elasticsearch 8 'username' = '$username', — 选填 用户名 'password' = '$password', — 选填 密码 'hosts' = 'http://xx.xx.xx.xx:9200', — Elasticsearch 的连接地址 'index' = 'my-index', — Elasticsearch 的 Index 名 'format' = 'json', — 输出数据格式，目前只支持 'json' 'sink.bulk-flush.max-actions' = '1000', — 每个批量请求的最大操作数 'sink.bulk-flush.interval' = '1s' — 批量 flush 的时间间隔);INSERT INTO elasticsearch8_sink_table SELECT * FROM datagen_source_table;

十八、作为维表关联查询（Elasticsearch 6/7 Lookup Join）

注意：说明：

注意：适用于 Flink 1.18 及以上版本。将 Elasticsearch 作为维表，与流表进行 JOIN 查询。

— 维表：Elasticsearch 中的字典表CREATE TABLE dim_es_user ( `id` INT, `name` STRING, `age` INT, PRIMARY KEY (`id`) NOT ENFORCED) WITH ( 'connector' = 'elasticsearch-7', — ES7 维表 'hosts' = 'http://xx.xx.xx.xx:9200', 'index' = 'user-dim', 'format' = 'json', 'lookup.cache' = 'PARTIAL', — 缓存策略：ALL / PARTIAL / NONE 'lookup.partial-cache.max-rows' = '10000', — PARTIAL 缓存最大行数 'lookup.partial-cache.expire-after-write' = '30min');– 流表：业务数据流CREATE TABLE kafka_order ( `order_id` INT, `user_id` INT, `amount` DOUBLE, `ts` TIMESTAMP(3)) WITH ( 'connector' = 'kafka', — … Kafka 连接配置 …);– 维表 JOIN 查询，用 user_id 关联 ES 维表中的用户信息INSERT INTO result_sinkSELECT o.order_id, o.amount, u.name, u.ageFROM kafka_order AS oLEFT JOIN dim_es_user FOR SYSTEM_TIME AS OF o.ts AS u ON o.user_id = u.id;

十九、作为批数据源

CREATE TABLE elasticsearch7_source_table ( `id` bigint, `event_date` int, `app` int, primary key (`id`) not enforced) with ( — 必填参数 'connector' = 'es-source', 'endPoint' = 'xx.xx.xx.xx', — Elasticsearch 的连接 ip 'accessId' = 'elastic', — 用户名 'accessKey' = 'PASSWORD', — 密码 'indexName' = 'my-index', — Elasticsearch 的 Index 名 'format' = 'json', — 数据格式，只支持 'json' — 可选参数 'scheme' = 'http', — 连接协议 'port' = '9200', — 端口 'batchSize' = '2000', — 每个 scroll 请求从 Elasticsearch 集群获取的最大文档数 'keepScrollAliveSecs' = '60' — scroll 上下文保留的最长时间，单位为分钟);CREATE TABLE logger_sink ( `id` bigint, `event_date` int, `app` int, primary key (`id`) not enforced) with ( 'connector' = 'logger');INSERT INTO logger_sink SELECT * from elasticsearch7_source_table;

20、注意事项

如果你希望连接其他版本的 Elasticsearch，请通过附加自定义程序包的方式，上传相应的 Elasticsearch Sink 的 JAR 包。

21、监控指标说明

Oceanus 为 ES Connector 增加了很多实用的统计指标。单击 Flink UI 的运行图中的 ES Sink 算子，即可搜索并查看指标：

numberOfInsertRecords：获取输出的 +I 消息数。

numberOfDeleteRecords：获取输出的 -D 消息数。

numberOfUpdateBeforeRecords：获取输出的 -U 消息数。

numberOfUpdateAfterRecords：获取输出的 +U 消息数。