FlinkCDC

Posted 2021-09-22 超级无敌小剑

 

1.CDC工具的种类

CDC主要分为基于查询和基于Binlog两种方式，这两种之间的区别：

	基于查询的CDC	基于Binlog的CDC
开源产品	Sqoop、Kafka JDBC Source	Canal、Maxwell、Debezium
执行模式	Batch	Streaming
是否可以捕获所有数据变化	否	是
延迟性	高延迟	低延迟
是否增加数据库压力	是	否

2.什么是FlinkCDC?

Flink社区开发了 flink-cdc-connectors 组件，这是一个可以直接从 mysql、PostgreSQL 等数据库直接读取全量数据和增量变更数据的 source 组件。目前也已开源，
FlinkCDC是基于Debezium的.

FlinkCDC相较于其他工具的优势:

1.能直接把数据捕获到Flink程序中当做流来处理,避免再过一次kafka等消息队列,而且支持历史数据同步,使用更方便.

2.FlinkCDC的断点续传功能:
 Flink-CDC将读取binlog的位置信息以状态的方式保存在CK,如果想要做到断点续传, 需要从Checkpoint或者Savepoint启动程序,通过这种方式来实现断点续传

3.FlinkCDC使用

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>Flink-CDC</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>1.12.0</version>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.12</artifactId>
            <version>1.12.0</version>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.12</artifactId>
            <version>1.12.0</version>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>3.1.3</version>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.48</version>
        </dependency>

        <dependency>
            <groupId>com.alibaba.ververica</groupId>
            <artifactId>flink-connector-mysql-cdc</artifactId>
            <version>1.2.0</version>
        </dependency>

        <!--cdc2.0-->
        <!--<dependency>
            <groupId>com.ververica</groupId>
            <artifactId>flink-connector-mysql-cdc</artifactId>
            <version>2.0.0</version>
        </dependency>-->

        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.75</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner-blink_2.12</artifactId>
            <version>1.12.0</version>
        </dependency>


    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>3.0.0</version>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>

3.1 FlinkCDC API使用

点击查看代码

package cdc;


import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;

import io.debezium.data.Envelope;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
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.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/*FlinkCDC 可以直接将mysql的binlog读取到Flink程序中 断点续传功能依赖于ck的保存 */
public class DataStreamAPITest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties debeProp = new Properties();
        // 配置 Debezium在初始化快照的时候（扫描历史数据的时候） =》 不要锁表
        debeProp.setProperty("debezium.snapshot.locking.mode", "none");

        env.setParallelism(1);
        System.setProperty("HADOOP_USER_NAME", "otto");
        /*//TODO 2.开启检查点   Flink-CDC将读取binlog的位置信息以状态的方式保存在CK,如果想要做到断点续传,
        // 需要从Checkpoint或者Savepoint启动程序

        //开启Checkpoint,每隔5秒钟做一次CK  ,并指定CK的一致性语义
        env.enableCheckpointing(TimeUnit.SECONDS.toMillis(5));

        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        //最大同时存在的ck数 和设置的间隔时间有一个就行
        checkpointConfig.setMaxConcurrentCheckpoints(1);
        //超时时间
        //checkpointConfig.setCheckpointTimeout(TimeUnit.SECONDS.toMillis(5));
        //2.3 指定从CK自动重启策略
        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, 5000L));
        //2.4 设置任务关闭的时候保留最后一次CK数据
        //checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        //设置状态后端
        env.setStateBackend(new FsStateBackend("hdfs://Ava01:8020/gmall/flinkcdc"));
        //env.setStateBackend(new FsStateBackend("file:///ck1/ck2",true));*/


        // TODO FlinkCDC的配置信息
        //DebeziumSourceFunction<String> MysqlSource = MySqlSource.<String>builder()
        DebeziumSourceFunction<String> MysqlSource = MySQLSource.<String>builder()
                .hostname("Ava01")
                .port(3306)
                .deserializer(new MyDeserializationSchema()) //去参数里面找找实现类
                .username("root")
                .password("123456")
                .databaseList("gmall_control") //可以指定多个库
                .tableList("gmall_control.table_process") //因为是多个库 所以要指定库名+表名
                .startupOptions(StartupOptions.initial())// 读取binlog策略 这个启动选项有五种
                .debeziumProperties(debeProp) //配置不要锁表 但是数据一致性不是精准一次 会变成最少一次
                .build();
        /*
         *  .startupOptions(StartupOptions.latest()) 参数配置
         *  1.initial() 全量扫描并且继续读取最新的binlog 最佳实践是第一次使用这个
         *  2.earliest() 从binlog的开头开始读取 就是啥时候开的binlog就从啥时候读
         *  3.latest() 从最新的binlog开始读取
         *  4.specificOffset(String specificOffsetFile, int specificOffsetPos) 指定offset读取
         *  5.timestamp(long startupTimestampMillis) 指定时间戳读取
         * */


        env.addSource(MysqlSource).print();

        env.execute("flink-cdc");


    }

    public static class MyDeserializationSchema implements DebeziumDeserializationSchema<String> {

        @Override //主要逻辑实现
        public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
            //从大的目标value里面将其他的Struct获取出来
            Struct value = (Struct) sourceRecord.value();
            Struct after = value.getStruct("after");
            Struct source = value.getStruct("source");

            String db = source.getString("db");//库名
            String table = source.getString("table");//表名

            //获取操作类型 直接将参数穿进去 会自己解析出来 里面是个enum对应每个操作
            /* READ("r"),
               CREATE("c"),
                UPDATE("u"),
                DELETE("d");*/
            Envelope.Operation operation = Envelope.operationFor(sourceRecord);
            String opstr = operation.toString().toLowerCase();
            //类型修正 会把insert识别成create
            if (opstr.equals("create")) {
                opstr = "insert";
            }

            //获取after结构体里面的表数据，封装成json输出
            JSONObject json1 = new JSONObject();
            JSONObject json2 = new JSONObject();
            //加个判空
            if (after != null) {
                List<Field> data = after.schema().fields(); //获取结构体
                for (Field field : data) {
                    String name = field.name(); //结构体的名字
                    Object value2 = after.get(field);//结构体的字段值
                    //放进json2里面去 json2放到json1里面去
                    json2.put(name, value2);
                }
            }


            //整理成大json串输出
            json1.put("db", db);
            json1.put("table", table);
            json1.put("data", json2);
            json1.put("type", opstr);

            collector.collect(json1.toJSONString());


        }

        @Override //指定返回的数据类型 flink框架有自己封装的一套类型
        public TypeInformation<String> getProducedType() {
            // return Types.STRING; //这两种一样
            return TypeInformation.of(String.class);
        }
    }
}

/*
 * FlinkCD环境
 * 使用fs(hdfs)保存ck 单机测试
 * 打开ck保存
 * 手动触发保存点
 * FLink savepoint jobid hdfs路径
 * 手动cancel job作业
 * 变更mysql数据 产生新的binlog
 * 指定从savepoint恢复job 查看能否断点续传
 * flink run  -s  保存点的hdfs路径 -c jar包
 * */

API支持自定义反序列化器:

自定义后的数据格式

==>得到轻量的 关键的数据
#需要的字段  
ConnectRecord{
    topic=\'mysql_binlog_source.gmall0408.z_user_info\', 
    value=Struct{
        after=Struct{id=1,name=zs},
        source=Struct{
            db=gmall0408,
            table=z_user_info,
        },
        op=c,
        ts_ms=1631585338506
    }
}

#序列化后 输出
{"data":{"name":"zs","id":"1"},"type":"isnert","db":"gmall_rt","table":"test_log"}
{"data":{"name":"xxx6","id":"5"},"type":"update","db":"gmall_rt","table":"test_log"}
{"data":{},"type":"delete","db":"gmall_rt","table":"test_log"}

3.2 FlinkCDC SQL应用

点击查看代码

package cdc;


import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.Properties;
/* 1.2 和 2.0 完全一致*/
public class SQLTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //


        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().useBlinkPlanner().build();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env, settings);

        //一个creat table 只能监控一个库里的一张表 但是api能一下监控多库多表
        String sql = "create table flink_cdc(id int,name String)\\n" +
                "WITH (\\n" +
                " \'connector\' = \'mysql-cdc\',\\n" +
                " \'hostname\' = \'Ava01\',\\n" +
                " \'port\' = \'3306\',\\n" +
                " \'username\' = \'root\',\\n" +
                " \'password\' = \'123456\',\\n" +
                " \'scan.startup.mode\'=\'initial\',"+ //设定启动模式 两种 none和latest-offset  别和debeziun.snapshot.mode同时指定
                " \'database-name\' = \'gmall_rt\',\\n" +
                " \'table-name\' = \'test_log\'\\n" +
                ")";
        tableEnvironment.executeSql(sql);
        tableEnvironment.executeSql("select * from flink_cdc").print();

        env.execute("flink-sql");
    }



}

1.x读取到的数据格式

1.x 读取到的数据格式
SourceRecord{
	sourcePartition={
		server=mysql_binlog_source
	},
	sourceOffset={
		file=mysql-bin.000001,
		pos=14491052,
		row=1,
		snapshot=true
	}
}ConnectRecord{
	topic=\'mysql_binlog_source.gmall_rt.test_log\',
	kafkaPartition=null,
	key=null,
	keySchema=null,
	value=Struct{
		after=Struct{
			id=1,
			name=zs
		},
		source=Struct{
			version=1.4.1.Final,
			connector=mysql,
			name=mysql_binlog_source,
			ts_ms=0,
			snapshot=true,
			db=gmall_rt,
			table=test_log,
			server_id=0,
			file=mysql-bin.000001,
			pos=14491052,
			row=0
		},
		op=c,
		ts_ms=1631616469264
	},
	valueSchema=Schema{
		mysql_binlog_source.gmall_rt.test_log.Envelope: STRUCT
	},
	timestamp=null,
	headers=ConnectHeaders(headers=)
}

3.3 FlinkCDC 断点续传的测试

注意:测试环境最好使用linux系统的jar提交 在idea上可能会出现ck保存失败问题
断点续传测试:
  
 1.自动保存的ck(关闭自动删除) 用ck启动
 2.使用手动的savepoint启动

因为设置的ck文件系统是hadoop 所以需要添加flink和hadoop的继承

在主机环境变量添加 
#FLINK集成HADOOP需要
export HADOOP_CLASSPATH=`hadoop classpath`
export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop
 
#source一下
source /etc/profile.d/my_ini.sh

#使用单机模式启动Flink
bin/start-cluster.sh
#提交任务给web
bin/flink run -m Ava01:8081 -c 全类名 jar包位置 -d 后台执行
bin/flink run -m Ava01:8081 -c com.otto.gmall.cdc.DataStreamAPITest myjar/Flink-CDC-1.0-SNAPSHOT-jar-with-dependencies.jar -d


#在Ava01:8081上查看webUI的输出
#手动触发保存点
flink savepoint jobId  sp保存在hdfs路径
 
在webUI上手动cancel job作业
变更监听的mysql库的表的数据 观察是否断点续传

#使用savepoint来恢复任务 观察能否断点续传
flink run  -s  保存点的hdfs路径  -c 全类名 jar包

3.4 FlinkCDC API和SQL的区别

1、sql的格式更轻量
2、api可以指定多库多表,sql一个建表语句只能指定一张表
3、动态分流的配置表：使用SQL的方式来同步

应用:动态分流 广播状态变量

3.5 FlinkCDC 1.x和2.x的区别

1. 依赖的区别： groupid没有了 alibaba
2. API写法区别： MySQLSource ---》 MySqlSource
3. sql写法一样 但是有兼容性问题 
   cdc2.0的SQL写法,只支持1.13以上的flink版本
   CDC2.0的sql语法只支持flink1.13版本
   最早的CDC，只支持flink1.11
4. FlinCDC 1.2.0 锁表机制

 Flink在读取bitlog的时候 为了保持数据一致性 会给该表添加一个全局读锁（只能查，不能增删改）
 然后记录bitlog 开始扫描全表 等待扫描完成之后 才会释放全局读锁,从记录的bitlog处开始同步 
 但是这个全局读锁的添加要求登录的mysql用户必须有RELOAD权限,否则添加的全局读锁会变成表级锁,在生产环境中会有大问题!

 解决办法: 使用API的时候  配置debeProp.setProperty("debezium.snapshot.locking.mode", "none"); 
                  让 Debezium在初始化快照的时候（扫描历史数据的时候）  不要锁表  但是会影响数据一致性,从精准一次性变成 最少一次

5. FlinkCDC 从2.x版本开始 修改了保持数据一致性的策略 不再锁表 而是采用水位的方式.

CDC2.0流程分析
无锁怎么保证一致性？
    一个SourceReader包含多个chunk
        一致性包括：
            一个chunk内部的一致性
            一个SourceReader里，多个chunk的一致性
    chunk读取：  记录binlog（低水位）
                =》 开始读取，得到数据 =》 放到一个buffer里（等待修正）
                =》 再次查看binlog（高水位）
                =》 如果 高水位 > 低水位，说明读取期间，有变化
                =》 获取变化的binlog，修正读取的数据
        =》单个Chunk中的数据一致性
        
    SourceReader内部多个Chunk的一致性：
        取多个chunk之间最大的 高水位 
        每个chunk去补足，自己的高水位到 最大高水位之间的数据
        ==》所有的chunk，都同步到了同一个进度

绝不摆烂