Flink 实践教程 - 入门（4）：读取 MySQL 数据写入到 ES

INSERT INTO `user` (`user_id`, `user_name`, `create_time`) VALUES (1001, '小明', '2021-10-01 00:00:00');INSERT INTO `user` (`user_id`, `user_name`, `create_time`) VALUES (1002, 'TONY', '2021-10-02 00:00:00');

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);