FlinkCDC

 

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，都同步到了同一個進度