CREATE TABLE `user_source` (
    `user_id` int,
    `user_name` varchar(50),
    PRIMARY KEY (`user_id`) NOT ENFORCED -- 如果要同步的数据库表定义了主键, 则这里也需要定义
) WITH (
    'connector' = 'mysql-cdc',      -- 必须为 'mysql-cdc'
    'hostname' = '10.0.0.158',      -- 数据库的 IP
    'port' = '3306',                -- 数据库的访问端口
    'username' = 'root',            -- 数据库访问的用户名（需要提供 SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT, SELECT, RELOAD 权限）
    'password' = 'yourpassword',    -- 数据库访问的密码
    'database-name' = 'testdb',     -- 需要同步的数据库
    'table-name' = 'user'           -- 需要同步的数据表名
);

-- Elasticsearch 只能作为数据目的表（Sink）写入
-- 参见 https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/connect.html#elasticsearch-connector

CREATE TABLE es_sink (
    `user_id` INT,
    `user_name` VARCHAR
) WITH (
    'connector.type' = 'elasticsearch',    -- 输出到 Elasticsearch
    'connector.version' = '6',             -- 指定 Elasticsearch 的版本, 例如 '6', '7'. 
    'connector.hosts' = 'http://10.0.0.175:9200', 
    'connector.index' = 'User',     
    'connector.document-type' = 'user',  
    'connector.username' = 'elastic',  
    'connector.password' = 'yourpassword', 

    'update-mode' = 'upsert',              -- 捕捉数据库变化时，需使用 'upsert' 模式 
    'connector.key-delimiter' = '$',       -- 可选参数, 复合主键的连接字符 (默认是 _ 符号)
    'connector.key-null-literal' = 'n/a',  -- 主键为 null 时的替代字符串，默认是 'null'
    'connector.connection-max-retry-timeout' = '300', -- 每次请求的最大超时时间 (ms)
    'format.type' = 'json'                 -- 输出数据格式, 目前只支持 'json'
);

insert into es_sink
(
    select user_id,
    LOWER(user_name) -- LOWER()函数会将用户名转换为小写
    from user_source
);