FlinkCDC的分析和应用代码

news2024/11/15 2:13:23

前言：原本想讲如何基于Flink实现定制化计算引擎的开发，并以FlinkCDC为例介绍；发现这两个在表达上不知以谁为主，所以先分析FlinkCDC的应用场景和技术实现原理，下一篇再去分析Flink能在哪些方面，做定制化计算引擎的开发操作。本文将从FlinkCDC应用场景开始，然后讲述其基于Flink的实现原理和代码应用，为下一篇介绍基于Flink开发定制化引擎做铺垫。

一、FlinkCDC应用场景

经常有同事或朋友问，Flink和FlinkCDC有什么区别？

Flink是一个流数据处理计算框架，FlinkCDC是数据采集工具：

Flink应用场景对比的是Storm、Spark；

FlinkCDC应用场景对比的是Sqoop、Canal、Maxwell和KafkaConnectSource、Debezium等；

FlinkCDC是Flink社区伙伴对数据采集需求，开发的一个SDK工具，让Flink在数据捕捉场景，使用起来更方便一些。

1.1 CDC的应用场景分析

       CDC的英文名是Change Data Capture (变化数据获取)；解决的应用场景，是对存储中间件中数据的采集，比如Mysql、Orcle、PGSql、MongoDB等中间件；

采集的方式分为基于查询和基于BinLog两种；

以mysql的数据采集为例：可以通过jdbc批次查询，也可以通过Binlog解析增量数据采集；

两者的一些特性对比如下：

基于查询的CDC直接获取数据，基于binlog的采集需要开启binlog服务。

1.2 FlinkCDC的应用分析

       FlinkCDC和Canal实现的应用场景需求是差不多的，都是通过binlog采集增量数据；

但是可用性上的不同是：  

对于cannal类似的采集服务需要三步:

  • 1.开启mysql的binlog
  • 2.将数据写到kafka
  • 3.用flink订阅kafka中的数据进行业务需求处理

对于FlinkCDC：只需要在binlog开启后，直接在一个Flink任务内做业务处理（可以写到kafka处理也行）；

       所有如cannal的采集功能服务，都需要单独维护一套服务，增加了运维负担，FlinkCDC可以当作任务部署到集群，大幅减轻了数据采集的应用难度；

用一个任务就完成这个应用功能：

二、FlinkCDC技术分析与本地操作

2.1 FlinkCDC的技术架构分析

       与Canal这些提供服务能力的服务不同，FlinkCdc只是一个任务，可以简单的开发和部署。

       Flink是借用了Debezium的功能，Debezium是一个可轻量级嵌入代码逻辑的服务，将Debezium的采集功能，用Flink的sourceFunction包装，然后打包成SDK提供给Flink开发使用；

       借用Flink自己的算子和sink能力，可以将采集到的数据以Flink的特性加工数据，并将数据写入Flink内置的connect组件，sink到服务里，如Kafka、Pulser、ES、RabbitMQ、MongoDB等。

2.2 本地操作
2.2.1准备mysql数据库表和数据
use flink_test;
#检测binlog是否开启
show variables like '%log_bin%'

#构建测试表
CREATE TABLE `event_info` (
  `id` int NOT NULL AUTO_INCREMENT,
  `name` varchar(255) NOT NULL,
  `category` varchar(512) DEFAULT NULL,
  `pv` int DEFAULT 0,
  `uv` int DEFAULT 0,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB  DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

#写入数据
insert  into `event_info`(`id`,`name`,`category`,`pv`,`uv`) values 
(1,'aaa','nfh',20,8),
(2,'bbb','dfgf',30,2),
(3,'ccc','fsd',40,4),
(4,'ddd','afs',50,7),
(5,'eee','asfa',60,3)
(6,'aaa','nfh',20,8),
(7,'bbb','dfgf',30,2),
(8,'ccc','fsd',40,4),
(9,'ddd','afs',50,7),
(10,'eee','asfa',60,3);
2.2.2 pom文件

       注意Flink和FlinkCDC的版本映射，很多显示的可以关联的版本之间是冲突的，这是一个很繁琐的工作，我调试各个版本之间的映射，花了一天左右的时间[求赞求收藏]；

下面这是Flink1.14.5版本和FlinkCDC2.2.1版本已经调好的依赖：

<?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>changedateDoris</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <scala.version>2.12</scala.version>
        <java.version>1.8</java.version>
        <flink.version>1.14.5</flink.version>
        <fastjson.version>1.2.62</fastjson.version>
        <hadoop.version>2.8.3</hadoop.version>
        <scope.mode>compile</scope.mode>
        <slf4j.version>1.7.30</slf4j.version>
    </properties>

    <dependencies>
        <!--        springboot 依赖-->
        <!-- flink  -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner_${scala.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_${scala.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_${scala.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-jdbc_${scala.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka_${scala.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>${fastjson.version}</version>
        </dependency>
        <!-- Add log dependencies when debugging locally -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <!-- mysql-connector -->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.12</version>
        </dependency>
        <dependency>
            <groupId>com.ververica</groupId>
            <artifactId>flink-sql-connector-mysql-cdc</artifactId>
            <version>2.2.1 </version>
            <exclusions>
                <exclusion>
                    <artifactId>flink-shaded-guava</artifactId>
                    <groupId>org.apache.flink</groupId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-runtime-web_${scala.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>

    </dependencies>

 <build>
<!--        <filters>-->
<!--            <filter>${project.basedir}/src/main/resources/env/application-${profileActive}.properties</filter>-->
<!--        </filters>-->
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>2.1</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <artifactSet>
                                <includes>
                                    <include>*:*</include>
                                </includes>
                                <excludes>
                                    <exclude>org.slf4j:slf4j-api:jar:</exclude>
                                </excludes>
                            </artifactSet>
                            <transformers>
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                                    <resource>META-INF/spring.handlers</resource>
                                </transformer>
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                                    <resource>META-INF/spring.schemas</resource>
                                </transformer>
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                                    <resource>META-INF/spring.factories</resource>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <artifactId>maven-resources-plugin</artifactId>
                <configuration>
                    <encoding>utf-8</encoding>
                    <useDefaultDelimiters>true</useDefaultDelimiters>
                    <delimiters>
                        <delimiter>$[*]</delimiter>
                    </delimiters>
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                        <configuration>
                            <descriptorRefs>
                                <descriptorRef>jar-with-dependencies</descriptorRef>
                            </descriptorRefs>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>8</source>
                    <target>8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>
2.2.3 java代码
package yto.com.net.demo;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class mysqlBinlogRead {
    
    private static final Logger log = LoggerFactory.getLogger(mysqlBinlogRead.class);

    public static void main(String[] args) throws Exception {

        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("ip")
                .port(3306)
                .databaseList("flink_test")
                .tableList("flink_test.event_info")
                .username("mysqlUser")
                .password("password")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
//                .startupOptions(StartupOptions.earliest())
                .build();

        Configuration configuration = new Configuration();
        configuration.setInteger(RestOptions.PORT, 8083);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // enable checkpoint
        env.enableCheckpointing(10000);

        DataStreamSource<String> cdcSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL CDC Source");

        SingleOutputStreamOperator<String> process = cdcSource.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String row, ProcessFunction<String, String>.Context context, Collector<String> collector) throws Exception {
                JSONObject rowJson = JSON.parseObject(row);

                String op = rowJson.getString("op");

                JSONObject source = rowJson.getJSONObject("source");
                String table = source.getString("table");

            }
        });
        
        cdcSource.print("message=:");
        env.execute("flinkCdc Read message");
    }


}
2.2.4 运行结果

如图所示，已经写入库的结果通过connect获取，增量数据通过binlog获取；

注意：表中的历史数据过多，全量读取的时候将会内存溢出。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1381497.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

案例109:基于微信小程序的高校寻物平台设计与实现

案例109:基于微信小程序的高校寻物平台设计与实现

文末获取源码 开发语言&#xff1a;Java 框架&#xff1a;SSM JDK版本&#xff1a;JDK1.8 数据库&#xff1a;mysql 5.7 开发软件&#xff1a;eclipse/myeclipse/idea Maven包&#xff1a;Maven3.5.4 小程序框架&#xff1a;uniapp 小程序开发软件&#xff1a;HBuilder X 小程序…
阅读更多...
SpringBoot之优化高并发场景下的HttpClient并提升QPS

SpringBoot之优化高并发场景下的HttpClient并提升QPS

HttpClient优化思路 使用连接池&#xff08;简单粗暴&#xff09; 长连接优化&#xff08;特殊业务场景&#xff09; httpclient和httpget复用 合理的配置参数&#xff08;最大并发请求数&#xff0c;各种超时时间&#xff0c;重试次数&#xff09; 异步请求优化&#xff0…
阅读更多...
【开源】基于JAVA+Vue+SpringBoot的校园电商物流云平台

【开源】基于JAVA+Vue+SpringBoot的校园电商物流云平台

目录 一、摘要1.1 项目介绍1.2 项目录屏 二、功能模块2.1 数据中心模块2.2 商品数据模块2.3 快递公司模块2.4 物流订单模块 三、系统设计3.1 用例设计3.2 数据库设计3.2.1 商品表3.2.2 快递公司表3.2.3 物流订单表 四、系统展示五、核心代码5.1 查询商品5.2 查询快递公司5.3 查…
阅读更多...
数据结构链表完整实现（负完整代码）

数据结构链表完整实现（负完整代码）

文章目录 前言引入1、链表定义及结构链表的分类3、单向不带头链表实现实现完整代码 4、带头双向循环链表实现实现完整代码 前言 引入 在上一篇文章中&#xff0c;我们认识了顺序表&#xff0c;但是在许多情况中&#xff0c;顺序表在处理一些事件时还存在许多问题&#xff0c;比…
阅读更多...
P1067 [NOIP2009 普及组] 多项式输出————C++

P1067 [NOIP2009 普及组] 多项式输出————C++

目录 [NOIP2009 普及组] 多项式输出题目描述输入格式输出格式样例 #1样例输入 #1样例输出 #1 样例 #2样例输入 #2样例输出 #2 提示 解题思路Code运行结果 [NOIP2009 普及组] 多项式输出 题目描述 一元 n n n 次多项式可用如下的表达式表示&#xff1a; f ( x ) a n x n a …
阅读更多...
现代 C++ 及 C++ 的演变

现代 C++ 及 C++ 的演变

C 活跃在程序设计领域。该语言写入了许多新项目&#xff0c;而且据 TIOBE 排行榜数据显示&#xff0c;C 的受欢迎度和使用率位居第 4&#xff0c;仅次于 Python、Java 和 C。 尽管 C 在过去二十年里的 TIOBE 排名都位居前列&#xff08;2008 年 2 月排在第 5 名&#xff0c;到…
阅读更多...
行为型设计模式——迭代器模式

行为型设计模式——迭代器模式

迭代器模式 迭代器模式也是非常的简单&#xff0c;定义如下&#xff1a; 提供一个对象来顺序访问聚合对象中的一系列数据&#xff0c;而不暴露聚合对象的内部表示。 相信大家都使用过类似下面的迭代器&#xff1a; List<String> list new ArrayList<>(); Iterat…
阅读更多...
Seata TC端协调全局事务

Seata TC端协调全局事务

1、Seata server注册器 //来自RM分支事务注册 super.registerProcessor(MessageType.TYPE_BRANCH_REGISTER, onRequestProcessor, messageExecutor); //开启全局事务 super.registerProcessor(MessageType.TYPE_GLOBAL_BEGIN, onRequestProcessor, messageExecutor); //提交全…
阅读更多...
MySQL夯实之路-事务详解

MySQL夯实之路-事务详解

事务四大特性 事务需要通过严格的acid测试。Acid表示原子性&#xff0c;一致性&#xff0c;隔离性&#xff0c;持久性。 原子性&#xff08;atomicity&#xff09; 事务是不可分割的最小单元&#xff0c;对于整个事务的操作&#xff0c;要么全部提交成功&#xff0c;要么全部…
阅读更多...
Linux———ps命令详解

Linux———ps命令详解

目录 ps 命令&#xff08;"process status" 的缩写。&#xff09; 常用选项和参数&#xff1a; a&#xff1a;显示所有用户的进程&#xff0c;包括其他用户的进程。​ u&#xff1a;显示详细的进程信息&#xff0c;包括进程的所有者、CPU 使用率、内存使用量等。…
阅读更多...
白嫖aws创建Joplin server服务器

白嫖aws创建Joplin server服务器

网上有很多的Joplin服务器的搭建教程&#xff0c;但是基本都是抄来抄去&#xff0c;对初学者实在是太不友好了。 话不多说&#xff0c;说干就干&#xff0c;自己从头找资料搭了一个&#xff0c;这可能是全网最好的Joplin服务器搭建教程了。 aws服务器 aws的服务器还是很香的&…
阅读更多...
MAVROS的进一步理解

MAVROS的进一步理解

一、Mavros简介 顾名思义&#xff0c; mavros就是mavlinkros。mavros是PX4官方提供的一个运行于ros下收发mavlink消息的工具&#xff0c;利用mavros可以发送mavlink消息给飞控(可以控制飞机)&#xff0c;并且可以从飞控中接受数据(例如&#xff1a;飞控的位置速度 IMU数据等等…
阅读更多...
论文阅读_训练大模型用于角色扮演

论文阅读_训练大模型用于角色扮演

英文名称: Character-LLM: A Trainable Agent for Role-Playing 中文名称: 角色-LLM&#xff1a;训练Agent用于角色扮演 文章: [https://arxiv.org/abs/2310.10158](https://arxiv.org/abs/2310.10158) 作者: Yunfan Shao, Linyang Li, Junqi Dai, Xipeng Qiu 机构: 复旦大学…
阅读更多...
使用numpy处理图片——分离通道

使用numpy处理图片——分离通道

大纲 读入图片分离通道堆叠法复制修改法 生成图片 在《使用numpy处理图片——滤镜》中&#xff0c;我们剥离了RGB中的一个颜色&#xff0c;达到一种滤镜的效果。 如果我们只保留一种元素&#xff0c;就可以做到PS中分离通道的效果。 读入图片 import numpy as np import PIL.…
阅读更多...
【UE Niagara学习笔记】04 - 火焰喷射时的黑烟效果

【UE Niagara学习笔记】04 - 火焰喷射时的黑烟效果

目录 效果 步骤 一、创建烟雾材质 二、添加新的发射器 三、设置新发射器 3.1 删除Color模块 3.2 减少生成的粒子数量 3.3 设置粒子初始颜色 3.4 设置烟雾的位置偏移 3.5 设置烟雾淡出 在上一篇博客&#xff08;【UE Niagara学习笔记】03 - 火焰喷射效果&#xf…
阅读更多...
【开源】基于JAVA+Vue+SpringBoot的医院门诊预约挂号系统

【开源】基于JAVA+Vue+SpringBoot的医院门诊预约挂号系统

目录 一、摘要1.1 项目介绍1.2 项目录屏 二、功能模块2.1 功能性需求2.1.1 数据中心模块2.1.2 科室医生档案模块2.1.3 预约挂号模块2.1.4 医院时政模块 2.2 可行性分析2.2.1 可靠性2.2.2 易用性2.2.3 维护性 三、数据库设计3.1 用户表3.2 科室档案表3.3 医生档案表3.4 医生放号…
阅读更多...
5文件操作

5文件操作

包含头文件<fstream> 操作文件三大类&#xff1a; ofstream : 写文件ifstream &#xff1a;读文件fstream : 读写文件 5.1文本文件 -文件以ascii的形式存储在计算机中 5.1.1写文件 步骤&#xff1a; 包含头文件 #include "fstream"创建流对象 ofs…
阅读更多...
【Linux运维】LVM和RAID学习及实践

【Linux运维】LVM和RAID学习及实践

LVM和RAID学习及实践 背景LVM简介新加硬盘的操作RAID-磁盘阵列应用场景RAID0RAID1其他结构RAID制作RAID 小结 背景 某台服务器的磁盘管理需要自己动手处理&#xff0c;找了一些资料也踩了一些坑&#xff0c;在这里记录一下&#xff0c;先介绍一下LVM和RAID这两个东西。在计算机…
阅读更多...
Java实现在线编辑预览office文档

Java实现在线编辑预览office文档

文章目录 1 在线编辑1.1 PageOffice简介1.2 前端项目1.2.1 配置1.2.2 页面部分 1.3 后端项目1.3.1 pom.xml1.3.2 添加配置1.3.3 controller 2 在线预览2.1 引言2.2 市面上现有的文件预览服务2.2.1 微软2.2.2 Google Drive查看器2.2.3 阿里云 IMM2.2.4 XDOC 文档预览2.2.5 Offic…
阅读更多...
逆变器3前级推免（高频变压器）

逆变器3前级推免（高频变压器）

一节电池标压是在2.8V—4.2V之间&#xff0c;所以24V电压需要大概七节电池串联。七节电池电压大概在19.6V—29.4V之间。 从24V的电池逆变到到220V需要升压的过程。那么我们具体需要升压到多少&#xff1f; 市电AC220V是有效值电压&#xff0c;峰值电压是220V*1.414311V 如果…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023