flink学习(13)—— 重试机制和维表join

重试机制

当任务出现异常的时候,会直接停止任务——解决方式,重试机制

1、设置checkpoint后,会给任务一个重启策略——无限重启

2、可以手动设置任务的重启策略

代码设置

//开启checkpoint后,默认是无限重启,可以设置该值 表示不重启
env.setRestartStrategy(RestartStrategies.noRestart());//作业失败flink中最多重启3次,每次重启的最小间隔是10s
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.of(10, TimeUnit.SECONDS)));//2分钟内最多重启3次,每次重启的最小间隔是5秒
env.setRestartStrategy(RestartStrategies.failureRateRestart(3,Time.of(2,TimeUnit.MINUTES),Time.of(5,TimeUnit.SECONDS))
);//无限重启
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(Integer.MAX_VALUE,  // 无限重启次数Time.of(10, TimeUnit.SECONDS)  // 每次重启的延迟时间
));

维表join

所谓的维表Join: 进入Flink的数据,需要关联另外一些存储设备的数据,才能计算出来结果

那么存储在外部设备上的表称之为维表,可能存储在mysql也可能存储在hbase 等。

维表一般的特点是变化比较慢。——名词表,维度表。

解决方式

 解决维表join的方式方式一:可以用一个静态代码块,或者在open方法中对一个集合初始化,用于存放想要相关联的数据。缺点:数据不能动态改变了方式二:在open中初始化连接,在map中每拿到流中的一条数据,就去mysql中查找一次缺点:数据可以动态改变,但是去mysql查找的次数太多了方式三:创建一个缓存区,用于存放数据,若过期则再去mysql中查询数据。没有缺点,可以动态获取数据了,也减少了mysql的查询次数(缓冲)唯一的是,若是多线程,可能会去mysql查询多次

方式一

package com.bigdata.day06;import com.mchange.v2.c3p0.ComboPooledDataSource;
import org.apache.commons.dbutils.QueryRunner;
import org.apache.commons.dbutils.handlers.MapHandler;
import org.apache.commons.dbutils.handlers.MapListHandler;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;import java.util.List;
import java.util.Map;
import java.util.Properties;/*** 直接从mysql中拿出* 弊端 只能拿到一次 不能实现动态*/
public class _03_维表join_01 {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();Properties properties = new Properties();properties.setProperty("bootstrap.servers", "bigdata01:9092");properties.setProperty("group.id", "g1");FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("edu",new SimpleStringSchema(),properties);DataStreamSource<String> source = env.addSource(consumer);source.map(new RichMapFunction<String, String>() {ComboPooledDataSource pool = null;QueryRunner queryRunner = null;List<Map<String, Object>> list = null;@Overridepublic void open(Configuration parameters) throws Exception {// 在open中执行sqlpool = new ComboPooledDataSource();queryRunner = new QueryRunner(pool);String sql = "select * from city ";list = queryRunner.query(sql, new MapListHandler());}@Overridepublic void close() throws Exception {pool.close();}@Overridepublic String map(String line) throws Exception {String[] split = line.split(",");Object cityName = "未知";for (Map<String, Object> map : list) {String cityId = (String)map.get("city_id");if (cityId.equals(split[1])){cityName = map.get("city_name");}}return line+","+cityName;}}).print();env.execute();}
}

方式二

package com.bigdata.day06;import com.mchange.v2.c3p0.ComboPooledDataSource;
import org.apache.commons.dbutils.QueryRunner;
import org.apache.commons.dbutils.handlers.MapHandler;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;import java.util.Map;
import java.util.Properties;/*** 每次从kafka中拿到一条数据就从mysql中查一遍* 弊端 对mysql的压力加大*/
public class _03_维表join_02 {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();Properties properties = new Properties();properties.setProperty("bootstrap.servers", "bigdata01:9092");properties.setProperty("group.id", "g1");FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("edu",new SimpleStringSchema(),properties);DataStreamSource<String> source = env.addSource(consumer);source.map(new RichMapFunction<String, String>() {ComboPooledDataSource pool = null;QueryRunner queryRunner = null;@Overridepublic void open(Configuration parameters) throws Exception {pool = new ComboPooledDataSource();queryRunner = new QueryRunner(pool);}@Overridepublic void close() throws Exception {pool.close();}@Overridepublic String map(String line) throws Exception {// 在处理逻辑中执行sqlString[] split = line.split(",");String sql = "select city_name from city where city_id = ?";Map<String, Object> rs = queryRunner.query(sql, new MapHandler(), split[1]);String cityName="未知";if (rs !=null){cityName = (String) rs.get("city_name");}return line+","+cityName;}}).print();env.execute();}
}

方式三

package com.bigdata.day06;import com.mchange.v2.c3p0.ComboPooledDataSource;
import org.apache.commons.dbutils.QueryRunner;
import org.apache.commons.dbutils.handlers.MapHandler;
import org.apache.commons.dbutils.handlers.MapListHandler;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.cache.*;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.TimeUnit;/*** 最终 非常好的方式* 现在内存中查 查不到在去mysql中找* 唯一的问题是,假如是多线程情况下,可能会触发多次去mysql中查找的方法*/
public class _03_维表join_03_cache {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();Properties properties = new Properties();properties.setProperty("bootstrap.servers", "bigdata01:9092");properties.setProperty("group.id", "g1");FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("edu",new SimpleStringSchema(),properties);DataStreamSource<String> source = env.addSource(consumer);// 记得设置并行度env.setParallelism(1);source.map(new RichMapFunction<String, String>() {ComboPooledDataSource pool = null;QueryRunner queryRunner = null;// 定义一个Cache// 第一个是传入的参数类型 第二个是存放的值的类型// 也就是,传入一个参数,根据这个值获取结果,拿的时候通过传入的值 拿存放的值LoadingCache<String, String> cache;@Overridepublic void open(Configuration parameters) throws Exception {pool = new ComboPooledDataSource();queryRunner = new QueryRunner(pool);cache = CacheBuilder.newBuilder()//最多缓存个数,超过了就根据最近最少使用算法来移除缓存 LRU.maximumSize(1000)//在更新后的指定时间后就回收// 不会自动调用,而是当过期后,又用到了过期的key值数据才会触发的。.expireAfterWrite(50, TimeUnit.SECONDS)//指定移除通知.removalListener(new RemovalListener<String, String>() {@Overridepublic void onRemoval(RemovalNotification<String, String> removalNotification) {System.out.println(removalNotification.getKey() + "被移除了,值为:" + removalNotification.getValue());}}).build(//指定加载缓存的逻辑new CacheLoader<String, String>() {// 假如缓存中没有数据,会触发该方法的执行,并将结果自动保存到缓存中@Overridepublic String load(String cityId) throws Exception {String sql = "select city_name from city where city_id = ? ";Map<String, Object> rs = queryRunner.query(sql, new MapHandler(), cityId);String cityName = null;if (rs!=null){cityName = (String) rs.get("city_name");}System.out.println("进入数据库查询成功,查询的值为"+cityId+"--"+cityName);return cityName;}});}@Overridepublic void close() throws Exception {pool.close();}@Overridepublic String map(String line) throws Exception {String[] arr = line.split(",");// 使用这种方式取值String cityName = cache.get(arr[1]);return line+","+cityName;}}).print();env.execute();}
}

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

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

相关文章

为什么爱用低秩矩阵

目录 为什么爱用低秩矩阵 一、定义与性质 二、区别与例子 为什么爱用低秩矩阵 我们更多地提及低秩分解而非满秩分解,主要是因为低秩分解在数据压缩、噪声去除、模型简化和特征提取等方面具有显著的优势。而满秩分解虽然能够保持数据的完整性,但在实际应用中的场景较为有限…

Dify+Docker

1. 获取代码 直接下载 &#xff08;1&#xff09;访问 langgenius/dify: Dify is an open-source LLM app development platform. Difys intuitive interface combines AI workflow, RAG pipeline, agent capabilities, model management, observability features and more, …

Android Studio的AI工具插件使用介绍

Android Studio的AI工具插件使用介绍 一、前言 Android Studio 的 AI 工具插件具有诸多重要作用&#xff0c;以下是一些常见的方面&#xff1a; 代码生成与自动补全 代码优化与重构 代码解读 学习与知识获取 智能搜索与资源推荐实际使用中可以添加注释&#xff0c;解读某段代…

DOCKER学习总结

这里写目录标题 一、Docker安装1.1 在线安装1.2 离线安装安装配置启动服务 1.3 配置镜像1.4 Docker启动相关命令 二、Docker三大核心概念2.1 镜像2.2 容器2.3 仓库2.3.1 公有仓库2.3.2 私有仓库 二、容器与虚拟机比较 一、Docker安装 1.1 在线安装 查看是否安装dockeryum lis…

深入浅出体验AI生图产品Dall-E

DALL-E是由OpenAI开发的一种革命性的AI图像生成工具&#xff0c;能够根据文本描述生成图像。它的名字灵感来源于著名画家萨尔瓦多达利&#xff08;Salvador Dal&#xff09;和皮克斯动画电影中的角色瓦力&#xff08;WALL-E&#xff09;&#xff0c;这暗示了其在艺术创造力与技…

OpenCV_Code_LOG

孔洞填充 void fillHole(const Mat srcBw, Mat &dstBw) {Size m_Size srcBw.size();Mat TempMat::zeros(m_Size.height2,m_Size.width2,srcBw.type());//延展图像srcBw.copyTo(Temp(Range(1, m_Size.height 1), Range(1, m_Size.width 1)));cv::floodFill(Temp, Point(…

YOLOv11改进,YOLOv11添加SAConv可切换空洞卷积,二次创新C3k2结构

摘要 作者提出的技术结合了递归特征金字塔和可切换空洞卷积,通过强化多尺度特征学习和自适应的空洞卷积,显著提升了目标检测的效果。 理论介绍 空洞卷积(Atrous Convolution)是一种可以在卷积操作中插入“空洞”来扩大感受野的技术,更有效地捕捉到图像中的大范围上下文…

2024信创数据库TOP30之华为Gauss DB

近日&#xff0c;由DBC联合CIW/CIS共同发布的“2024信创数据库TOP30”榜单正式揭晓&#xff0c;汇聚了国内顶尖的数据库企业及其产品&#xff0c;成为展示中国信创领域技术实力与发展潜力的重要平台。在这份榜单中&#xff0c;华为的GaussDB凭借其卓越的技术实力、广泛的行业应…

【Spring源码核心篇-07】spring事物传播机制的流程和原理

Spring源码核心篇整体栏目 内容链接地址【一】Spring的bean的生命周期https://zhenghuisheng.blog.csdn.net/article/details/143441012【二】深入理解spring的依赖注入和属性填充https://zhenghuisheng.blog.csdn.net/article/details/143854482【三】精通spring的aop的底层原…

Redis实现限量优惠券的秒杀

核心&#xff1a;避免超卖问题&#xff0c;保证一人一单 业务逻辑 代码步骤分析 全部代码 Service public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {Resourceprivate ISeckillVoucher…

.NET8/.NETCore 依赖注入:自动注入项目中所有接口和自定义类

.NET8/.NETCore 依赖接口注入&#xff1a;自动注入项目中所有接口和自定义类 目录 自定义依赖接口扩展类&#xff1a;HostExtensions AddInjectionServices方法GlobalAssemblies 全局静态类测试 自定义依赖接口 需要依赖注入的类必须实现以下接口。 C# /// <summary>…

搭建一个基于Web的文档管理系统,用于存储、共享和协作编辑文档

搭建一个基于Web的文档管理系统&#xff0c;用于存储、共享和协作编辑文档 本项目采用以下架构&#xff1a; NFS服务器: 负责存储文档资料。Web服务器: 负责提供文档访问和编辑功能。SELinux: 负责权限控制&#xff0c;确保文档安全。Git服务器: 负责存储文档版本历史&#x…

gitee:创建仓库,存入本地文件至仓库

一、git下载 git:下载与安装-CSDN博客https://blog.csdn.net/weixin_46001736/article/details/144107485?sharetypeblogdetail&sharerId144107485&sharereferPC&sharesourceweixin_46001736&spm1011.2480.3001.8118 二、创建仓库 1、主页面->右上角新增…

计算机网络 —— HTTP 协议(详解)

前一篇文章&#xff1a;网页版五子棋—— WebSocket 协议_网页可以实现websocket吗-CSDN博客 目录 前言 一、HTTP 协议简介 二、HTTP 协议格式 1.抓包工具的使用 2.抓包工具的原理 3.抓包结果 4.HTTP协议格式总结 三、HTTP 请求 1. URL &#xff08;1&#xff09;UR…

关于单片机的原理与应用!

成长路上不孤单&#x1f60a;&#x1f60a;&#x1f60a;&#x1f60a;&#x1f60a;&#x1f60a; 【14后&#x1f60a;///计算机爱好者&#x1f60a;///目前正在学习C&#x1f60a;///持续分享所学&#x1f60a;///如有需要欢迎收藏转发///&#x1f60a;】 今日分享关于单片…

爬虫专栏第一篇:深入探索爬虫世界:基础原理、类型特点与规范要点全解析

本专栏会对爬虫进行从0开始的讲解&#xff0c;每一步都十分的细致&#xff0c;如果你感兴趣希望多多点赞收藏关注支持 简介&#xff1a;文章对爬虫展开多方面剖析。起始于爬虫的基本概念&#xff0c;即依特定规则在网络抓取信息的程序或脚本&#xff0c;在搜索引擎信息提取上作…

rabbitmq原理及命令

目录 一、RabbitMQ原理1、交换机&#xff08;Exchange&#xff09;fanoutdirecttopicheaders&#xff08;很少用到&#xff09; 2、队列Queue3、Virtual Hosts4、基础对象 二、RabbitMQ的一些基本操作:1、用户管理2、用户角色3、vhost4、开启web管理接口5、批量删除队列 一、Ra…

@antv/x6 再vue中 ,自定义图形,画流程图、数据建模、er图等图形

X6 是基于 HTML 和 SVG 的图编辑引擎&#xff0c;提供低成本的定制能力和开箱即用的内置扩展&#xff0c;方便我们快速搭建 DAG 图、ER 图、流程图、血缘图等应用。 最终效果图 1.安装 npm install antv/x6 --save //x6主要包 npm install antv/x6-vue-shape //使用vue组…

vscode + conda + qt联合开发

安装vscode 安装conda 清华大学开源软件镜像(Anaconda下载)_清华大学镜像-CSDN博客 conda create新建一个环境&#xff0c;激活这个环境&#xff0c;然后安装pyside6 pip install pyside6 -i https://pypi.tuna.tsinghua.edu.cn/simple 安装成功后输入 pip list查看是否安装…

debian 11 虚拟机环境搭建过坑记录

目录 安装过程系统配置修改 sudoers 文件网络配置换源安装桌面mount nfs 挂载安装复制功能tab 无法补全其他安装 软件配置eclipse 配置git 配置老虚拟机硬盘挂载 参考 原来去 debian 官网下载了一个最新的 debian 12&#xff0c;安装后出现包依赖问题&#xff0c;搞了半天&…