FlinkCDC快速搭建实现数据监控

引入依赖

<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>com.sand</groupId><artifactId>flinkcdc</artifactId><version>1.0-SNAPSHOT</version><packaging>jar</packaging><name>Flink Quickstart Job</name><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><flink.version>1.17.1</flink.version><!--        <flink.version>1.14.4</flink.version>--><target.java.version>1.8</target.java.version><scala.binary.version>2.12</scala.binary.version><maven.compiler.source>${target.java.version}</maven.compiler.source><maven.compiler.target>${target.java.version}</maven.compiler.target><log4j.version>2.17.1</log4j.version></properties><repositories><repository><id>apache.snapshots</id><name>Apache Development Snapshot Repository</name><url>https://repository.apache.org/content/repositories/snapshots/</url><releases><enabled>false</enabled></releases><snapshots><enabled>true</enabled></snapshots></repository></repositories><dependencies><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><!--            <artifactId>flink-streaming-java_2.12</artifactId>--><artifactId>flink-streaming-java</artifactId><version>${flink.version}</version></dependency><!--        <dependency>--><!--            <groupId>org.apache.flink</groupId>--><!--            <artifactId>flink-scala_2.12</artifactId>--><!--            <version>${flink.version}</version>--><!--        </dependency>--><dependency><groupId>org.apache.flink</groupId><!--            <artifactId>flink-clients_2.12</artifactId>--><artifactId>flink-clients</artifactId><version>${flink.version}</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>5.1.49</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-core</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-planner_2.12</artifactId><version>${flink.version}</version></dependency><!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-elasticsearch7 --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-elasticsearch7</artifactId><version>3.0.1-1.17</version></dependency><dependency><groupId>com.ververica</groupId><artifactId>flink-connector-mysql-cdc</artifactId><!-- The dependency is available only for stable releases, SNAPSHOT dependency need build by yourself. --><version>2.4.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-json</artifactId><version>${flink.version}</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.75</version></dependency><!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-jdbc --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-jdbc</artifactId><version>3.1.1-1.17</version><scope>provided</scope></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>2.0.32</version></dependency><!-- 打印日志的jar包 --><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>1.7.30</version></dependency><dependency><groupId>log4j</groupId><artifactId>log4j</artifactId><version>1.2.16</version></dependency><dependency><groupId>org.apache.logging.log4j</groupId><artifactId>log4j-core</artifactId><version>2.17.2</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.30</version></dependency><dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId><version>5.7.10</version></dependency></dependencies><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.1</version><configuration><source>${target.java.version}</source><target>${target.java.version}</target></configuration></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-shade-plugin</artifactId><version>3.1.1</version><executions><!-- Run shade goal on package phase --><execution><phase>package</phase><goals><goal>shade</goal></goals><configuration><createDependencyReducedPom>false</createDependencyReducedPom><artifactSet><excludes><exclude>org.apache.flink:flink-shaded-force-shading</exclude><exclude>com.google.code.findbugs:jsr305</exclude><exclude>org.slf4j:*</exclude><exclude>org.apache.logging.log4j:*</exclude></excludes></artifactSet><filters><filter><artifact>*:*</artifact><excludes><exclude>META-INF/*.SF</exclude><exclude>META-INF/*.DSA</exclude><exclude>META-INF/*.RSA</exclude></excludes></filter></filters><transformers><transformerimplementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/><transformerimplementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"><mainClass>com.sand.DataStreamJob</mainClass></transformer></transformers></configuration></execution></executions></plugin></plugins><pluginManagement><plugins><!-- This improves the out-of-the-box experience in Eclipse by resolving some warnings. --><plugin><groupId>org.eclipse.m2e</groupId><artifactId>lifecycle-mapping</artifactId><version>1.0.0</version><configuration><lifecycleMappingMetadata><pluginExecutions><pluginExecution><pluginExecutionFilter><groupId>org.apache.maven.plugins</groupId><artifactId>maven-shade-plugin</artifactId><versionRange>[3.1.1,)</versionRange><goals><goal>shade</goal></goals></pluginExecutionFilter><action><ignore/></action></pluginExecution><pluginExecution><pluginExecutionFilter><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><versionRange>[3.1,)</versionRange><goals><goal>testCompile</goal><goal>compile</goal></goals></pluginExecutionFilter><action><ignore/></action></pluginExecution></pluginExecutions></lifecycleMappingMetadata></configuration></plugin></plugins></pluginManagement></build>
</project>

数据库配置类

package com.sand;import org.apache.commons.collections.CollectionUtils;import java.util.Arrays;
import java.util.List;
import java.util.StringJoiner;/*** @author zdd*/
public class CDCKit {public static void main(String[] args) {String tempDir = System.getProperty("java.io.tmpdir");System.out.println("tempDir = " + tempDir);}/*** 数据库*/private static final String database = "byyy_iowtb_wms_test";/*** 表名*/private static final List<String> tableList = Arrays.asList("inv_tt_stock_info","base_tm_sku","base_tm_third_sku_certificate","base_tm_sku_gsp");/*** ip*/private static final String hostname = "192.168.111.107";/*** 端口*/private static final int port = 3306;/*** 用户名*/private static final String username = "test_cdc";/*** 密码*/private static final String password = "Test_cdc@123";public static String getDatabase() {return database;}public static String getTableList() {if (CollectionUtils.isEmpty(tableList)) {return null;}//,分割StringJoiner stringJoiner = new StringJoiner(",");for (String tableName : tableList) {stringJoiner.add(getDatabase() + "." + tableName);}return stringJoiner.toString();}public static String getHostname() {return hostname;}public static int getPort() {return port;}public static String getUsername() {return username;}public static String getPassword() {return password;}}

监控类

package com.sand;import cn.hutool.core.io.FileUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.io.File;
import java.util.Objects;
import java.util.Properties;public class DataStreamJob {public static void main(String[] args) throws Exception {//获取临时文件目录String tempDir = System.getProperty("java.io.tmpdir");String latestCheckpoint = getLatestCheckpoint();System.out.println("latestCheckpoint = " + latestCheckpoint);Configuration configuration = new Configuration();if(StringUtils.isNotBlank(latestCheckpoint)){configuration.setString("execution.savepoint.path", "file:///" + latestCheckpoint);}StreamExecutionEnvironment env =StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);env.setParallelism(1);//2.1 开启 Checkpoint,每隔 60 秒钟做一次 CKenv.enableCheckpointing(1000L * 60);//2.2 指定 CK 的一致性语义env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);//2.3 设置任务关闭的时候保留最后一次 CK 数据env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);//2.4 指定从 CK 自动重启策略env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000L));//2.5 设置状态后端env.setStateBackend(new FsStateBackend("file:///" + tempDir + "ck"));// ck 设置env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);Properties properties = new Properties();properties.setProperty("snapshot.locking.mode", "none");properties.setProperty("decimal.handling.mode", "string");MySqlSource<String> sourceFunction = MySqlSource.<String>builder().hostname(CDCKit.getHostname()).port(CDCKit.getPort()).databaseList(CDCKit.getDatabase()).tableList(CDCKit.getTableList()).username(CDCKit.getUsername()).password(CDCKit.getPassword()).scanNewlyAddedTableEnabled(true).deserializer(new JsonDebeziumDeserializationSchema()).startupOptions(StartupOptions.initial()).debeziumProperties(properties).build();//4.使用 CDC Source 从 MySQL 读取数据env.fromSource(sourceFunction, WatermarkStrategy.noWatermarks(), "mysql-source").addSink(new MysqlSink());//5.打印数据
//        mysqlStream.print();//6.执行任务env.execute();}private static String getLatestCheckpoint() {File ckDir = new File(System.getProperty("java.io.tmpdir") + "ck");File[] files = ckDir.listFiles();if (files == null) {return null;}String path = null;long lastModified = 0;for (File file : files) {//获取文件夹下-chk-开头文件夹-最新的文件夹if (file.isDirectory()) {File[] files1 = file.listFiles();if (files1 == null) {continue;}for (File file1 : files1) {if (!file1.isDirectory() || !file1.getName().startsWith("chk-")) {continue;}if (file1.lastModified() > lastModified) {lastModified = file1.lastModified();path = file1.getAbsolutePath();}}}}//删除其余目录if (StringUtils.isEmpty(path)) {return null;}String tempPath = path.substring(0, path.lastIndexOf("\\"));for (File file : files) {if (file.isDirectory() && !Objects.equals(file.getAbsolutePath(), tempPath)) {FileUtil.del(file);}}return path;}
}

数据处理类

package com.sand;/*** @author zdd*/
public class MysqlSink implements org.apache.flink.streaming.api.functions.sink.SinkFunction<String> {@Overridepublic void invoke(String value, org.apache.flink.streaming.api.functions.sink.SinkFunction.Context context) throws Exception {System.out.println("value = " + value);}
}

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

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

相关文章

MySQL 针对逗号拼接的数据字段转行思路

MySQL 针对逗号拼接的数据字段转行思路

一、MySQL 针对逗号拼接的数据字段转行思路 在 MySQL 中我们有可能为了方便操作&#xff0c;有时会将一个字段存储多个信息&#xff0c;使用英文逗号隔开&#xff0c;当然这种情况属于对数据库的设计上有些欠妥。但如果遇到了这种情况又需要对数据进行统计的情况就有点棘手了&…
阅读更多...
golang常见面试题

golang常见面试题

1. go语言有哪些优点、特性&#xff1f; 语法简便&#xff0c;容易上手。 支持高并发&#xff0c;go有独特的协程概念&#xff0c;一般语言最小的执行单位是线程&#xff0c;go语言支持多开协程&#xff0c;协程是用户态线程&#xff0c;协程的占用内存更少&#xff0c;协程只…
阅读更多...
STM32CubeIDE基础学习-设置输出HEX和BIN文件的方法

STM32CubeIDE基础学习-设置输出HEX和BIN文件的方法

STM32CubeIDE基础学习-设置输出HEX和BIN文件的方法 前言 当某些时候&#xff0c;需要把一个程序下载到单片机里面进行功能验证或者看结果时&#xff0c;可以用串口软件来烧录程序&#xff0c;而不用再打开程序工程在线烧录程序这么麻烦了&#xff0c;那么就需要该工程文件夹下…
阅读更多...
电脑记事本怎么查看字数 记事本字数便捷查看方法

电脑记事本怎么查看字数 记事本字数便捷查看方法

在数字化的时代&#xff0c;电脑记事本已成为我记录生活、工作的得力助手。相较于传统的纸质笔记本&#xff0c;它的便捷性不言而喻&#xff1a;随时随地&#xff0c;打开就能写&#xff0c;无需担心纸张用尽或笔墨不干的尴尬。但有一个问题一直困扰着我&#xff0c;那就是如何…
阅读更多...
UI学习 一

UI学习 一

教程&#xff1a;Accessibility – Material Design 3 需要科学上网&#xff0c;否则图片显示不出来。设计教程没有图片说明&#xff0c;不容易理解。 优化UI方向 清晰可见的元素足够的对比度和尺寸重要性的明确等级一眼就能辨别的关键信息 传达某一事物的相对重要性 将重…
阅读更多...
Centos7 安装mongoDB

Centos7 安装mongoDB

下载安装包 curl -O https://fastdl.mongodb.org/linux/mongodb-linux-x86_64-3.2.12.tgz 解压 tar -zxvf mongodb-linux-x86_64-3.2.12.tgz移动到指定位置 mv mongodb-linux-x86_64-3.2.12/ /usr/local/mongodb在/usr/local/mongodb下创建文件夹 cd /usr/local/mongodb m…
阅读更多...
（二十二）从零开始搭建k8s集群——高可用kubernates集群搭建上篇

（二十二）从零开始搭建k8s集群——高可用kubernates集群搭建上篇

前言 本节内容分为上、中、下三篇&#xff0c;上篇主要是关于搭建k8s的基础环境&#xff0c;包括服务器基本环境的配置&#xff08;网络、端口、主机名、防火墙、交换分区、文件句柄数等&#xff09;、docker环境部署安装配置、镜像源配置等。中篇会介绍k8s的核心组件安装、k8…
阅读更多...
xss.pwnfunction.com靶机 Warmups

xss.pwnfunction.com靶机 Warmups

通关要求弹出警告框alert(1337) 没有用户交互 不能使用外链接 在chrome中测试 Ma Spaghet! 通过分析代码我们可以看到它直接用innerHTML将接收的内容赋值 但是我们不能使用<script>标签因为&#xff1a;HTML 5 中指定不执行由 innerHTML 插入的 <script> 标签。 所…
阅读更多...
网络原理初识

网络原理初识

一、IP地址 概念 IP 地址主要用于标识网络主机、其他网络设备&#xff08;如路由器&#xff09;的网络地址。简单说&#xff0c; IP 地址用于定位主机 的网络地址 。 就像我们发送快递一样&#xff0c;需要知道对方的收货地址&#xff0c;快递员才能将包裹送到目的地。 二、…
阅读更多...
YOLOv8原创改进：原创自研head创新 | 空间上下文感知模块（SCAM）结合超轻量高效动态上采样DySample | 小目标涨点系列

YOLOv8原创改进：原创自研head创新 | 空间上下文感知模块（SCAM）结合超轻量高效动态上采样DySample | 小目标涨点系列

💡💡💡本文独家改进：YOLOV8 head创新，1）一种超轻量高效动态上采样DySample， 具有更少的参数、FLOPs，效果秒杀CAFFE和YOLOv8网络中的nn.Upsample；2）加入空间上下文感知模块（SCAM）进一步提升检测精度； 改进结构图如下： 💡💡💡在多个数据集下验证能够涨点…
阅读更多...
使用Unreal Engine 5创建逼真人物化身(meta human)的教程

使用Unreal Engine 5创建逼真人物化身(meta human)的教程

这是一个关于使用Unreal Engine 5创建逼真人物化身(meta human)的教程视频。作者们使用PolyCam应用程序对一位名叫Angelina的女性进行了3D人体扫描,然后在Blender中清理扫描数据,并将其导入Unreal Engine 5。接着他们利用MetaHuman Creator网站进行了进一步的定制,将Angelina的…
阅读更多...
Java中SpringBoot四大核心组件是什么

Java中SpringBoot四大核心组件是什么

一、Spring Boot Starter 1.1 Starter的应用示例 <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-thymeleaf</artifactId> </dependency> <dependency><groupId>org.mybatis.sprin…
阅读更多...
Java开发从入门到精通（一）：Dubbo

Java开发从入门到精通（一）：Dubbo

目录&#xff1a; 服务注册与发现机制 服务调用 线程派发机制 1、服务注册与发现机制 Dubbo的服务注册与发现机制如下图所示&#xff1a; Dubbo中存在4类角色&#xff1a; Registry 注册中心。 Consumer 服务调用者、消费端。 Provider 服务提供者。 Monitor 监控中心。 调用…
阅读更多...
docker学习笔记——对数据卷的一些简单命令

docker学习笔记——对数据卷的一些简单命令

高级数据卷 docker volume ls&#xff1a;查看docker所有维护的数据卷。 docker volume inspect 数据卷别名&#xff1a;查看数据卷的详细信息。 docker volume rm 数据卷的别名&#xff1a;删除数据卷。 docker volume create 数据卷别名&#xff1a;创建一个别名数据卷。
阅读更多...
Android屏幕共享-硬编码硬解码

Android屏幕共享-硬编码硬解码

Android屏幕共享-硬编码硬解码 说起Android之间的屏幕共享&#xff0c;第一次接触会比较陌生&#xff0c;不过大家多少有了解过ffmpeg&#xff0c;看上去是不是很熟悉&#xff1f;ffmpeg是一套处理音视频的开源程序&#xff0c;但对于C了解较少的同学&#xff0c;编译起来很复杂…
阅读更多...
笔试题7——在数组中找出三个数字和为0的全部组合， test ok

笔试题7——在数组中找出三个数字和为0的全部组合， test ok

题目描述： 在数组中找出三个数字使得它们的和为0，找出所有的这种不包含重复的组合 思路： 首先将数组元素从小到大排序，设三个元素下标i，j，k，将i从头开始遍历，j初始化为i+1，k初始化为n-1； while(j < k) 当三元素之和小于0时，j++； 当三元素之和大于0时，k–； 当三…
阅读更多...
二维码门楼牌管理系统应用场景：社会服务与福利的智能化革新

二维码门楼牌管理系统应用场景：社会服务与福利的智能化革新

文章目录 前言一、二维码门楼牌管理系统的基本功能二、在社会服务领域的应用三、在福利保障领域的应用四、结论 前言 在数字化浪潮的推动下&#xff0c;二维码门楼牌管理系统以其便捷、高效的特点&#xff0c;正逐渐渗透到社会服务和福利的各个领域。这一系统不仅为市民提供了…
阅读更多...
Claude3 目前最强大模型

Claude3 目前最强大模型

过内试用连接 ChatAI (whjin.cn)http://chatyy.whjin.cn/?userid51913030&typebaidu&keywordid0&bd_vid12153741466503666331 Claude 3: A new generation of AI The Claude 3 family of models represents the cutting edge of AI technology, offering unpara…
阅读更多...
RN开发搬砖经验之-如何处理FlashList组件加载后调用scrollToIndex没有滚动指定位置

RN开发搬砖经验之-如何处理FlashList组件加载后调用scrollToIndex没有滚动指定位置

前言 如题&#xff0c;这里只能说是处理&#xff0c;起正向作用的临时方案&#xff0c;因为我也着实没搞懂这个BUG的具体原因&#xff0c;看github上有提相关的issuesFor long lists with different item types scrollToIndex does not work reliable&#xff0c;但看官方没有…
阅读更多...
Java的基本数据类型——四类八种

Java的基本数据类型——四类八种

目录 前言 1.整型 1.1整型变量 1.2长整型变量 1.3短整型变量 1.4字节型变量 2.浮点型 2.1双精度浮点数 2.2单精度浮点数 3.字符型 4.布尔型 前言 java的数据类型分为基本数据类型与引用数据类型,其中除了整型、浮点型、字符型、布尔型&#xff0c;即为引用数据类型…
阅读更多...
最新文章

Copyright @ 2022~2023