心得技巧
html5HTML/XhtmlCSSXML/XSLTDreamweaver教程Frontpage教程心得技巧

FlinkCDC

发布时间：2022-07-05 发布网站：脚本宝典
脚本宝典收集整理的这篇文章主要介绍了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.@R_406_2358@e.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支持自定义反序列化器:

@H_571_126@ 自定义后的数据格式 
==>得到轻量的 关键的数据
#需要的字段  
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没有了 alibaba2. API写法区别： MySQLSource ---》 MySqlSource3. sql写法一样 但是有兼容性问题    cdc2.0的SQL写法,只支持1.13以上的flink版本   CDC2.0的sql语法只支持flink1.13版本   最早的CDC，只支持flink1.114. 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，都同步到了同一个进度

脚本宝典总结

以上是脚本宝典为你收集整理的FlinkCDC全部内容，希望文章能够帮你解决FlinkCDC所遇到的问题。

如果觉得脚本宝典网站内容还不错，欢迎将脚本宝典推荐好友。

本图文内容来源于网友网络收集整理提供，作为学习参考使用，版权属于原作者。
如您有任何意见或建议可联系处理。小编QQ：384754419，请注明来意。
标签：batbathadoophdfsjavalinuxlistloadloadmysqlmysqlpartitionphppythonString一致性参数参数工具工具序列化序列化开发开发数据库数据库程序员程序员
猜你在找的心得技巧相关文章
其他相关热搜词更多
phpjavapython程序员loadpost-format-gallerydivString参数开发ListcapMapClass工具安装SQL数据库资源this
全站导航更多
html5HTML/XhtmlCSSXML/XSLTDreamweaver教程Frontpage教程心得技巧JavaScriptASP.NETPHP编程正则表达式AJAX相关ASP编程JSP编程编程10000问CSS/HTMLFlexvbsDOS/BAThtahtcpythonperl游戏相关VBA远程脚本ColdFusionMsSqlMysqlmariadboracleDB2mssql2008mssql2005SQLitePostgreSQLMongoDB星外虚拟主机华众虚拟主机Linuxwin服务器FTP服务器DNS服务器Tomcatnginxzabbix云和虚拟化bios系统安装系统系统进程Windows系列LINUXRedHat/CentosUbuntu/DebianFedoraSolaris红旗Linux建站经验微信营销网站优化网站策划网络赚钱网络创业站长故事alexa域名photoshop教程摄影教程Fireworks教程CorelDraw教程Illustrator教程Painter教程Freehand教程IndesignSketch笔记本主板内存CPU存储显卡显示器光存储鼠标键盘平板电脑安全教程杀毒防毒安全设置病毒查杀脚本攻防入侵防御工具使用业界动态Exploit漏洞分析
最新心得技巧教程
热门心得技巧教程