ICode9

精准搜索请尝试: 精确搜索
首页 > 数据库> 文章详细

flink cdc sql 开发模板,及踩坑记录

2021-10-29 16:02:24  阅读:74  来源: 互联网

标签:STRING cdc mysql flink api sql import


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包冲突

标签:STRING,cdc,mysql,flink,api,sql,import
来源: https://blog.csdn.net/WuBoooo/article/details/121036556

专注分享技术,共同学习,共同进步。侵权联系[admin#icode9.com]

Copyright (C)ICode9.com, All Rights Reserved.

ICode9版权所有