flink cdc sql 开发模板

flink cdc sql 读mysql的binlog日志，实时同步到mysql开发模板

使用flink cdc前提条件：读取目标库的用户必须开启binlog权限

package com.ysservice;
 
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import com.ysservice.utils.MyCheckpoint;
import com.ysservice.utils.SystemConstants;
import com.ysservice.yszt.owner.yszt_owner_customer_base_info;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.log4j.Logger;
 
import static org.apache.flink.api.common.time.Time.seconds;
 
/**
 * @Description:用flink cdc同步mysql数据
 * @author: WuBo
 * @date:2021/10/19 15:21
 */
public class TestDemo {
 
    public static void main(String[] args) throws Exception {
 
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //创建tableEnv
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //开启Checkpoint
        env.enableCheckpointing(60*1000);//开启chechPoint，每60秒记录一次中间状态
        env.getCheckpointConfig().setCheckpointTimeout(60*1000);//记录状态的超时时间为60秒
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(10);//chechPoint最多失败次数，因为Flink CDC Connector 在初始的全量快照同步阶段，会屏蔽掉快照的执行
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);//保存状态的类型的精准一次
        env.setRestartStrategy(RestartStrategies.failureRateRestart(5, seconds(60), seconds(2)));//60秒内报错5次，终止程序，每次重启间隔2秒
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);//停止任务时，保留Checkpoint
 
        //创建flink cdc的输入表,  datatime 的字段类型要改成 timestamp,否则会有时区问题
        tableEnv.executeSql("CREATE TABLE Data_Input (" +
                " ID bigint," +                                                             //字段类型
                " PROJECT_ID bigint," +                                                     //字段类型
                " PROJECT_CODE STRING," +                                                   //字段类型
                " PROJECT_NAME STRING," +                                                   //字段类型
                " AMOUNT decimal(20,2)," +                                                  //字段类型
                " ACTUAL_TYPE STRING," +                                                    //字段类型
                " TYPE_NAME STRING," +                                                      //字段类型
                " CREATED_AT timestamp," +                                                  //字段类型
                " CREATED_MAN STRING," +                                                    //字段类型
                " UPDATED_AT timestamp," +                                                  //字段类型
                " UPDATED_MAN STRING," +                                                    //字段类型
                " PRIMARY KEY (`ID`) NOT ENFORCED " +                                       //mysql表的主键，这个必须设置，否则不能无锁分布式读取和切块
                ") WITH (" +
                " 'connector' = 'mysql-cdc'," +                                             //connector类型：mysql-cdc
                " 'hostname' = '"+ SystemConstants.dataInput_hostname_test +"'," +          //MySQL的hostname，此处用的配置文件获取
                " 'port' = '3306'," +
                " 'username' = '"+ SystemConstants.dataInput_username_test +"'," +          //MySQL的username，此处用的配置文件获取
                " 'password' = '"+ SystemConstants.dataInput_password_test +"'," +          //MySQL的password，此处用的配置文件获取
                " 'database-name' = 'test'," +                                              //要读取的库名
                " 'table-name' = 'OUT_NORM_RULE_LIBRARY'," +                                //要读取的表名
                //" 'scan.startup.mode' = 'latest-offset'," +
                " 'scan.incremental.snapshot.enabled' = 'true'" +                           //增量式快照启动，启用后可以无锁分布式读表，默认启用
                ")");
 
        //创建输出表
        tableEnv.executeSql("CREATE TABLE Data_Output (" +
                " ID bigint," +
                " PROJECT_ID bigint," +
                " PROJECT_CODE STRING," +
                " PROJECT_NAME STRING," +
                " AMOUNT decimal(20,2)," +
                " ACTUAL_TYPE STRING," +
                " TYPE_NAME STRING," +
                " CREATED_AT timestamp," +
                " CREATED_MAN STRING," +
                " UPDATED_AT timestamp," +
                " UPDATED_MAN STRING," +
                " PRIMARY KEY (`ID`) NOT ENFORCED " +
                ") WITH (" +
                " 'connector' = 'jdbc'," +                                                   //输出表使用jdbc connector输出到mysql
                " 'url' = '"+ SystemConstants.dataOutput_url_datapush_out +"'," +
                " 'username' = '"+ SystemConstants.dataOutput_username_datapush_out +"'," +
                " 'password' = '"+ SystemConstants.dataOutput_password_datapush_out +"'," +
                " 'table-name' = 'OUT_NORM_RULE_LIBRARY2'" +
                ")");
        //执行sql，执行sql时，flink会自动判断过来的数据是插入还是删除（updata会变成两条数据，先删除再插入），并且会自动判断主键是否已经存在，存在就upsert
        tableEnv.executeSql("INSERT INTO Data_Output (SELECT * FROM Data_Input)");
 
    }
}

flink cdc 踩坑记录：

以下总结都是基于flink 1.13.2 对应的 flink cdc 2.0的

1. flink cdc 分两种api代码，一种是datastream api，一种是sql api，两种api有较大的差异，在这总结一下两种api的优劣势：

datastream api优势：可以读多库多表，代码灵活

                    劣势：只能单并行度读表，且mysql的datatime类型和timestamp的数据读出来有时区问题，而且程序启动时，需要reload锁表权限去做全量快照，会短暂的锁表，而且不能做Checkpoint

sql api 优势：可以多并行度的读表，且不需要锁表，定义数据类型时将datatime定义为timestamp类型，也能避免时区的问题，还能做Checkpoint

                    劣势：只能读取单表

2.datastream api作业在扫描 MySQL 全量数据时，checkpoint 超时，出现作业 failover

原因：Flink CDC 在 scan 全表数据，而在 scan 全表过程中是没有 offset 可以记录的（意味着没法做 checkpoint），但是 Flink 框架任何时候都会按照固定间隔时间做 checkpoint，所以此处 mysql-cdc source 做了比较取巧的方式，即在 scan 全表的过程中，会让执行中的 checkpoint 一直等待甚至超时。超时的 checkpoint 会被仍未认为是 failed checkpoint，默认配置下，这会触发 Flink 的 failover 机制，而默认的 failover 机制是不重启。所以会造成上面的现象

解决办法：配置 failed checkpoint 容忍次数，以及失败重启策略

3.datastream api执行时报锁权限问题

原因： 由于使用的 mysql 用户未授权 RELOAD 权限，导致无法获取全局读锁（FLUSH TABLES WITH READ LOCK）， CDC source 就会退化成表级读锁，而使用表级读锁需要等到全表 scan 完，才能释放锁，所以会发现持锁时间过长的现象，影响其他业务写入数据。

解决方法：给使用的 MySQL 用户授予 RELOAD 权限即可

4.sql api 正常提交任务后，只读全量数据，不读增量数据

原因：sql api在分布式全量读表完成后需要做一次全量的checkpoint，因为checkpoint未开启，导致无法进行下一步读取增量数据

解决方法：开启checkpoint还有输入表和输出表的binlog权限

5.mysql的datatime和timestamp数据类型时区问题

在使用datastream api读出来的datatime类型数据，会将年月日的数据类型读成时间戳的类型，那是因为binlog在存储datatime数据类型时，就是用时间戳的形式存储的，且该时间搓有时区问题，和现实时间差8小时，timestamp类型的数据读出来虽然不是时间戳类型的，但是依然会有8小时的时区差异，所以在使用datastream api时需要手动进行时区转换（datastream api目前没有找到其他解决方案）

但使用sql api时，读取datatime类型的数据时，只需要将该字段类型定义为timestamp去读取，就能解决时区和时间戳的问题，timestamp类型的数据正常读取即可，但是在使用sql api写入mysql时，需要在输出库中配置一下时区为+8:00，避免写入时造成时区问题，否则时间会相差12-13小时

6.运行flink任务时，flink输出的日志为空

原因：log4j jar包冲突

解决方法：将项目的log4j依赖全部排除掉，因为flink有自带的log4j jar包，我们再上传log4j jar包很容易造成jar包冲突