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

CREATE TABLE elasticsearch8_lookup_table (
    `id` INT,
    `name` STRING,
    PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
    'connector' = 'elasticsearch-8', -- ES8 维表
    'username' = '$username', -- 选填 用户名
    'password' = '$password', -- 选填 密码
    'hosts' = 'http://10.28.28.94:9200', -- Elasticsearch 的连接地址
    'index' = 'my-index', -- Elasticsearch 的 Index 名
    'format' = 'json', -- 输出数据格式，目前只支持 'json'
    'lookup.cache' = 'PARTIAL', -- 维表缓存策略，可选 ALL / PARTIAL / NONE
    'lookup.partial-cache.max-rows' = '10000', -- PARTIAL 缓存最大行数
    'lookup.partial-cache.expire-after-access' = '30min', -- PARTIAL 缓存访问过期时间
    'lookup.partial-cache.expire-after-write' = '30min' -- PARTIAL 缓存写入过期时间
);

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 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://10.28.28.94: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 中的字典表
CREATE TABLE dim_es_user (
    `id` INT,
    `name` STRING,
    `age` INT,
    PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
    'connector' = 'elasticsearch-8',            -- ES8 维表
    'hosts' = 'http://10.28.28.94: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_sink
SELECT
    o.order_id,
    o.amount,
    u.name,
    u.age
FROM kafka_order AS o
LEFT 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' = '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;