Flinkcdc监测mysql数据库，自定义反序列化，利用flinkApi和flinkSql两种方式

//创建流式环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//创建流式表环境
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

#### 第二步：配置mysql

> 
> 去mysql中的my.cnf中更改配置，作者的my.cnf是在/etc目录下面的，如果去进行修改可能权限不够
> 
> 
> 命令：sudo vim /etc/my.cnf
> 
> 
> 

![](https://img-blog.csdnimg.cn/direct/c50d7bc58dfc4b2dacfe38c953ee57f4.png)

以上图片的四个配置缺一不可，框起来的一定要注意，binlog - do -db=目标数据库名 

**更改完之后要重启mysql，这一步很重要，否则还是之前的配置**

> 
>  重启命令：sudo systemctl restart mysqld  
>  查看数据库命令：sudo systemctl status mysqld
> 
> 
> 

![](https://img-blog.csdnimg.cn/direct/d7ea30083d5f453f965208c750c22c63.png)

有箭头指示就表示完成了 

####  第三步：在目标数据库中创建需要检测的表

![](https://img-blog.csdnimg.cn/direct/602ad5989289483d854ef129319a33a5.png)

并在表中插入数据

#### 第四步：去flink官网找到连接mysql-cdc的语法格式

官网网址：[MySQL | Apache Flink CDC]( )

![](https://img-blog.csdnimg.cn/direct/e3a57a1a1e0e40b38d2f0fe84eeb7c77.png)

上面是我从flink官网上截下来的，我们只需要将官网上的内容复制，自己去修改就可以了 

tEnv.executeSql(“CREATE TABLE Person (\n” +
" id INT,\n"+
" name STRING,\n" +
" age BIGINT,\n" +
" sex String,\n" +
" PRIMARY KEY (id) NOT ENFORCED\n"+ --一定要加上主键，否则就会报错
“) WITH (\n” +
" ‘connector’ = ‘mysql-cdc’,\n" + --必填
" ‘hostname’=‘hadoop102’,\n" + --主机名，必填
" ‘port’=‘3306’,\n" + --端口号，必填
" ‘username’ = ‘root’,\n" + --用户名，必填
" ‘password’=‘123456’,\n" + --密码，必填
" ‘database-name’=‘zg3’,\n" + --目标数据库名，必填
" ‘table-name’=‘Person’,\n" + --目标表名，必填
" ‘jdbc.properties.useSSL’=‘false’\n" + --官网上可以找到，必填
“);”);

以上 是检测mysql中信息改的设置 

 报错

![](https://img-blog.csdnimg.cn/direct/d0ce33c8a2c14fb7bfebe18b26a3f31e.png)

> 
> Caused by: org.apache.flink.table.api.ValidationException: Multiple factories for identifier 'mysql-cdc' that implement 'org.apache.flink.table.factories.DynamicTableFactory' found in the classpath.  
>   
> 
> 
> 

如果你报这种错，就要看一下你自己的pom文件 

 ![](https://img-blog.csdnimg.cn/direct/32bd93a519664af68d36ae4a635dd074.png)

**这两个依赖冲突了**

 ![](https://img-blog.csdnimg.cn/direct/90b0fe2370c64d7a81673ea513ca023d.png)

这样就可以解决报错了 

![](https://img-blog.csdnimg.cn/direct/e952702421514c3ba661f46f8bec0e8f.png)

对刚才数据进行修改名字

![](https://img-blog.csdnimg.cn/direct/b7cb45ec4edc467a8eaa7f061722db53.png)

 ![](https://img-blog.csdnimg.cn/direct/c7e50b22ca584d51b7b61254a2073cdb.png)

**如果修改数据库，但是idea更新不出来值，这可能是好多人电脑的问题**

 ![](https://img-blog.csdnimg.cn/direct/96e93d07b85e4d549270cc48ddc70619.png)

**这时候将并行度添加为1就可以了**

#### 第五步：将以上数据以json格式传送到kafka中

官网网址：[Debezium | Apache Flink]( )

 Debezium-json:是Debezium为变更日志提供的统一格式结构。它是一种特定格式的JSON，用于描述数据库中的变更事件，如插入、更新或删除操作。这些事件可以由Debezium从源数据库捕获，并转换为JSON格式的消息，然后发布到消息队列（如Kafka）或对象存储（如OSS）中

 ![](https://img-blog.csdnimg.cn/direct/bde1c0df59fe4dceafa50ef881cb6631.png)

tEnv.executeSql(“CREATE TABLE kfk_cdc (\n” +
" id INT,\n" +
" name STRING,\n" +
" age BIGINT,\n" +
" sex String,\n" +
" PRIMARY KEY (id) NOT ENFORCED\n"+
“) WITH (\n” +
" ‘connector’ = ‘kafka’,\n" +
" ‘topic’ = ‘mysql_cdc’,\n" + --写到kafka的主题，必填
" ‘properties.bootstrap.servers’ = ‘hadoop102:9092’,\n" + --kafka服务器名
" ‘properties.group.id’ = ‘testGroup’,\n" +
" ‘scan.startup.mode’ = ‘earliest-offset’,\n" +
" ‘format’ = ‘debezium-json’\n" + --日志json格式，必填
“)”);

tEnv.executeSql(“insert into kfk_cdc select * from Person”);

 最后将目标表中的数据写到kafka的表中，因为flinksql连接kafka底层会将kafka表中数据自动写到kafka中

 ![](https://img-blog.csdnimg.cn/direct/a13d8a7fce3845ccb441c58c7fd6b1b2.png)

可能这样你们看不清，我把json数据美化一下 

![](https://img-blog.csdnimg.cn/direct/27d5f431d7ad4511b9e15e729c8a9b32.png)

### 第二种用FlinkApi实现

第二步，第三步和FlinkSql一样

第四步从官网上复制: [Overview | Apache Flink CDC]( )

![](https://img-blog.csdnimg.cn/direct/7f76624ab1744a9e90e3ca77dc03358b.png)

//免密配置
Properties properties = new Properties();
properties.put(“useSSL”,“false”);
MySqlSource mySqlSource = MySqlSource.builder()
.hostname(“hadoop102”) --必填 主机名
.port(3306) --必填 端口号
.databaseList(“zg3”) // set captured database --必填 数据库
.tableList(“zg3.Person”)// set captured table --必填 数据表
.username(“root”) --必填 用户名
.password(“123456”) --必填 密码
.deserializer(new JsonDebeziumDeserializationSchema()) --反序列化器
.jdbcProperties(properties) --jdbc配置，用来写免密配置
.build();

####  第五步：获取上一步源数据

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    // enable checkpoint
    env.enableCheckpointing(3000);

    env
            .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
            // set 4 parallel source tasks
            .setParallelism(1)
            .print(); // use parallelism 1 for sink to keep message ordering