最新版Flink CDC MySQL同步Elasticsearch（一）

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代码，如下所示：

CREATETABLE products (
 id INTNOTNULL,
 name STRING,
 description STRING,PRIMARYKEY(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>CREATETABLE products (>     id INT,>     name STRING,>     description STRING,>PRIMARYKEY(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代码，如下所示：

CREATETABLE product1 (
    id INT,
    name STRING,
    description STRING,PRIMARYKEY(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 SQLCREATETABLE product1 (>     id INT,>     name STRING,>     description STRING,>PRIMARYKEY(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>insertinto 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第一篇讲解完毕，希望能帮助到搭建