1.环境准备
首先我们要基于Flink CDC MySQL同步MySQL的环境基础上（flink-1.17.1、Java8、MySQL8）搭建Elasticsearch7-17-10和Kibana 7.17.10。笔者已经搭建好环境，这里不做具体演示了，如果需要Es的搭建教程情况笔者其他博客
注意： 建议生产环境统一使用稳定版本Flink1.16.*。笔者这里只是作为教程编写采用当下最新版本，生产环境不推荐使用
2.编译flink-sql-connector-mysql-cdc
最新版本flink-1.17.1 mysql同步Es具体jar依赖版本如下所示：
注意：下载链接仅适用于稳定版本，SNAPSHOT依赖需要您自己构建。
flink-sql-connector-elasticsearch7-3.0.1-1.17.jar
flink-sql-connector-mysql-cdc-2.5-SNAPSHOT.jar（需要自行进行构建编译，笔者构建的已经上次至次博客。需要可以进行下载，csdn需要积分下载，无法设置免费的，需要免费版可以直接联系笔者）
下载所需的JAR包并放在下面flink-1.17.1/lib/：
git clone https://github.com/ververica/flink-cdc-connectors.git
cd flink-cdc-connectors
mvn clean install -DskipTests
3.建立mysql和Es映射关系表
使用以下命令启动 Flink SQL CLI：
./bin/sql-client.sh
我们应该看到 CLI 客户端的欢迎屏幕。
首先，每 3 秒启用一次检查点
-- Flink SQL
Flink SQL> SET execution.checkpointing.interval = 3s;
编辑源数据库Flink Sql代码，如下所示：
CREATE TABLE products (
id INT NOT NULL,
name STRING,
description STRING,
PRIMARY KEY(id) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc', #引入的CDC jar包驱动，没有引入会报错提示需要引入
'hostname' = '192.168.50.163',#源数据库连接host地址，可以根据自己的具体设置，此处为笔者本机的
'port' = '3306', #源数据库端口
'username' = 'root',#源数据库账号
'password' = '*****',#源数据库密码
'database-name' = 'mydb',#源数据库
'table-name' = 'products'#源数据库表
);
在Flink SQL 执行以下语句创建从相应数据库表捕获更改数据的表
-- Flink SQL
Flink SQL> CREATE TABLE products (
> id INT,
> name STRING,
> description STRING,
> PRIMARY KEY (id) NOT ENFORCED
> ) WITH (
> 'connector' = 'mysql-cdc',
> 'hostname' = '192.168.50.163',
> 'port' = '3306',
> 'username' = 'root',
> 'password' = '****',
> 'database-name' = 'mydb',
> 'table-name' = 'products'
> );
在es创建要同步的目标索引，具体语句如下：
PUT product1
{
"settings": {
"number_of_shards": 12,
"number_of_replicas": 0
},
"mappings": {
"properties": {
"id": {
"type": "integer"
},
"name": {
"type": "keyword"
},
"description": {
"type": "text"
}
}
}
}
编辑目标ES映射Flink Sql代码，如下所示：
CREATE TABLE product1 (
id INT,
name STRING,
description STRING,
PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'elasticsearch-7',#目标ES版本，最新目前支持7
'hosts' = 'http://192.168.50.236:9200',#连接信息
'index' = 'product1'#索引信息
);
注意： 本文Es为测试版本没有配置账号密码，如果有账号密码配置即可 ‘username’ = ‘xxxx’,‘password’=‘xxxx’
建立目标索引与Flink SQL的映射关系，具体语句如下：
-- Flink SQL
CREATE TABLE product1 (
> id INT,
> name STRING,
> description STRING,
> PRIMARY KEY (id) NOT ENFORCED
> ) WITH (
> 'connector' = 'elasticsearch-7',#目标ES版本，最新目前支持7
> 'hosts' = 'http://192.168.50.236:9200',#连接信息
> 'index' = 'product1'#索引信息
> );
使用Flink SQL添加mysql和Es映射表数据关联关系
-- Flink SQL
Flink SQL> insert into product1 select * from products;
4.时区问题处理
错误：
The MySQL server has a timezone offset (28800 seconds ahead of UTC) which does not match the configured timezone Etc/UTC. Specify the right server-time-zone to avoid inconsistencies for time-related fields.
解决思路：
- Flink集群开启NTP服务器 时间同步
- 把服务器时区改成和数据库一样的时间本文为（Asia/Shanghai）
- 配置Flink sql的时区为Asia/Shanghai，具体命令如下所示：
Flink SQL> SET 'table.local-time-zone' = 'Asia/Shanghai';
注意：这是笔者遇到的问题，具体问题具体解决即可
5.具体实现结果
整体实现结果如下图所示：
Flink 运行任务
mysql 源数据表数据
Es目标索引已经数据查询图
至此，笔者的Flink CDC MySQL同步Elasticsearch第一篇讲解完毕，希望能帮助到搭建
