数据分析引擎 Elasticsearch Service

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://10.28.28.94:9200', -- Elasticsearch 的连接地址
    'index' = 'my-index', -- Elasticsearch 的 Index 名
    'document-type' = '_doc', -- Elasticsearch 的 Document 类型
    'format' = 'json' -- 输出数据格式，目前只支持 'json'
);

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://10.28.28.94:9200', -- Elasticsearch 的连接地址
    'index' = 'my-index', -- Elasticsearch 的 Index 名
    'format' = 'json' -- 输出数据格式，目前只支持 'json'
);

CREATE TABLE elasticsearch7_source_table (
  `id` bigint,
  `event_date` int,
  `app` int,
  primary key (`id`) not enforced
) with (
  -- 必填参数
  'connector' = 'es-source',
  'endPoint' = '127.0.0.1', -- 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 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://10.28.28.94: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;

CREATE TABLE elasticsearch7_source_table (
  `id` bigint,
  `event_date` int,
  `app` int,
  primary key (`id`) not enforced
) with (
  -- 必填参数
  'connector' = 'es-source',
  'endPoint' = '127.0.0.1', -- 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;