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

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第一篇讲解完毕，希望能帮助到搭建