Flink 实践教程：入门（2）：写入 Elasticsearch

-- Datagen Connector 可以随机生成一些数据用于测试-- 参见 https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/table/connectors/datagen.html
CREATE TABLE random_source (   f_sequence INT,   f_random INT,   f_random_str VARCHAR   ) WITH (   'connector' = 'datagen',   'rows-per-second'='1',  -- 每秒产生的数据条数        'fields.f_sequence.kind'='sequence',   -- 有界序列（结束后自动停止输出）  'fields.f_sequence.start'='1',         -- 序列的起始值  'fields.f_sequence.end'='10000',       -- 序列的终止值        'fields.f_random.kind'='random',       -- 无界的随机数  'fields.f_random.min'='1',             -- 随机数的最小值  'fields.f_random.max'='1000',          -- 随机数的最大值        'fields.f_random_str.length'='10'      -- 随机字符串的长度);

-- Elasticsearch 只能作为数据目的表（Sink）写入-- 注意! 如果您启用了 Elasticsearch 的用户名密码鉴权功能, 目前只能使用 Flink 1.10 的旧语法。若无需鉴权, 则可以使用 Flink 1.11 的新语法。-- 参见 https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/connect.html#elasticsearch-connector
CREATE TABLE Student (    `user_id`   INT,    `user_name` VARCHAR) WITH (    'connector.type' = 'elasticsearch', -- 输出到 Elasticsearch
    'connector.version' = '6',            -- 指定 Elasticsearch 的版本, 例如 '6', '7'. 注意务必要和所选的内置 Connector 版本一致    'connector.hosts' = 'http://10.0.0.175:9200',  -- Elasticsearch 的连接地址    'connector.index' = 'Student',        -- Elasticsearch 的 Index 名    'connector.document-type' = 'stu',    -- Elasticsearch 的 Document 类型    'connector.username' = 'elastic',     -- 可选参数: 请替换为实际 Elasticsearch 用户名    'connector.password' = 'xxxxxxxxxx',  -- 可选参数: 请替换为实际 Elasticsearch 密码
    'update-mode' = 'append',             -- 可选无主键的 'append' 模式，或有主键的 'upsert' 模式         'connector.key-delimiter' = '$',      -- 可选参数, 复合主键的连接字符 (默认是 _ 符号, 例如 key1_key2_key3)    'connector.key-null-literal' = 'n/a',  -- 主键为 null 时的替代字符串，默认是 'null'    'connector.failure-handler' = 'retry-rejected',   -- 可选的错误处理。可选择 'fail' （抛出异常）、'ignore'（忽略任何错误）、'retry-rejected'（重试）
    'connector.flush-on-checkpoint' = 'true',   -- 可选参数, 快照时不允许批量写入（flush）, 默认为 true    'connector.bulk-flush.max-actions' = '42',  -- 可选参数, 每批次最多的条数    'connector.bulk-flush.max-size' = '42 mb',  -- 可选参数, 每批次的累计最大大小 (只支持 mb)    'connector.bulk-flush.interval' = '60000',  -- 可选参数, 批量写入的间隔 (ms)    'connector.connection-max-retry-timeout' = '300',     -- 每次请求的最大超时时间 (ms)
    'format.type' = 'json'        -- 输出数据格式, 目前只支持 'json');

INSERT INTO StudentSELECTf_sequence   AS user_id,f_random_str AS user_nameFROM random_source;