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

?作者：腾讯云流计算 Oceanus 团队流计算 Oceanus 简介流计算 Oceanus 是大数据产品生态体系的实时化分析利器，是基于 Apache Flink 构建的具备一站开发、无缝连接、亚秒延时、低廉成本、安全稳定等特点的企业级实时大数据分析平台。流计算 Oceanus 以实现企业数据价值最大化为目标，加速企业实时化数字化的建设进程。本文将为您详细介绍如何使用 MySQL 接入数据，经过流计算 Oceanus 对数据进行处理分析（示例中采用小写转换函数对name字段进行了小写转换），最终将处理好的数据存入 Elasticsearch 中。
操作视频

前置准备
1.MySQL 集群准备 1.1 新建 MySQL 集群进入 MySQL 控制台[1]，点击左上方【新建】创建集群。具体可参考官方文档创建 mysql 实例[2]。在【数据库管理】> 【参数设置】中设置参数 binlog_row_image=FULL，便于使用 CDC（Capture Data Change）特性，实现数据的变更实时捕获。
1.2 准备数据首先创建 testdb 库，并在 testdb 库中创建用户 user 表，并插入数据。user 表结构：

字段名	类型	含义
user_id	int	用户ID
user_name	varchar(50)	用户名
create_time	timestamp	创建时间

在表中插入2条数据。

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

1.3 设置参数点击实例 ID，在实例详情页面点击【数据库管理】进入【参数设置】面板，设置binlog_row_image=FULL来开启数据库变化的同步。

通过 MySQL 集成数据到流计算 Oceanus (Flink) 集群，可以使用 flink-connector-jdbc 或者 flink-connector-mysq-cdc。使用 MySQL-cdc 特性时，flink-connector-mysq-cdc 连接器需要设置 MySQL 数据库的参数 binlog_row_image=FULL。

2. 创建流计算 Oceanus 集群
进入流计算 Oceanus 控制台[3]，点击左侧【集群管理】，点击左上方【创建集群】，具体可参考流计算 Oceanus 官方文档创建独享集群[4]。

创建流计算 Oceanus 集群和 MySQL 集群时所选 VPC 必须是同一 VPC。

3. 创建 Elasticsearch 集群
进入 Elasticsearch 控制台[5]，点击左上方【新建】，创建 Elasticsearch 实例，具体操作请访问创建 Elasticsearch 集群[6]。

创建 ES 集群和流计算 Oceanus 集群时所选私有网络 VPC 必须是同一 VPC。

流计算 Oceanus 作业 1. 创建 Source

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'-- 需要同步的数据表名 );

2. 创建 Sink

-- Elasticsearch 只能作为数据目的表（Sink）写入 -- 参见 https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/connect.html#elasticsearch-connectorCREATE 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' );

3. 编写业务 SQL

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

4. 选择 Connector
点击【保存】>【发布草稿】运行作业。

请根据实际购买的 Elasticsearch 版本选择对应的 Connector ，1.13 版本之后无需选择可自动匹配 Connector。

5. 数据查询
进入 Elasticsearch 控制台[5]，点击之前购买的 Elasticsearch 实例，点击右上角【Kibana】，进入 Kibana 查询数据。具体查询方法请参考通过 Kibana 访问集群[7]。
总结本示例用 MySQL 连接器持续集成数据库数据变化记录，经过流计算 Oceanus 实现最基础的数据转换功能，最后 Sink 到Elasticsearch 中，用户无需提前在 Elasticsearch 中创建索引。另外，ES 作为Source/Sink ，使用时间戳 timestamp 类型字段时长度需指定，如：timestamp(3)
参考阅读 [1]: MySQL 控制台：https://console.cloud.tencent.com/cdb
[2]: 创建 mysql 实例：https://cloud.tencent.com/document/product/236/46433
[3]: 流计算 Oceanus 控制台：https://console.cloud.tencent.com/oceanus/overview
[4]: 创建 Oceanus 独享集群：https://cloud.tencent.com/document/product/849/48298
[5]: Elasticsearch 控制台：https://console.cloud.tencent.com/es
[6]: 创建 Elasticsearch 集群：https://cloud.tencent.com/document/product/845/19536
[7]: 通过 Kibana 访问集群：https://cloud.tencent.com/document/product/845/19541