flink学习(6)——自定义source和kafka
概述
SourceFunction:非并行数据源(并行度只能=1) --接口
RichSourceFunction:多功能非并行数据源(并行度只能=1) --类
ParallelSourceFunction:并行数据源(并行度能够>=1) --接口
RichParallelSourceFunction:多功能并行数据源(并行度能够>=1) --类 【建议使用的】
——Rich 字样代表富有,在编程中,富有代表可以调用的方法很多,功能很全的意思。
基础案例
package com.bigdata.day02;//1、SourceFunction
// public class ZidingyiSource implements SourceFunction<Student> {
//2、RichSourceFunction
// public class ZidingyiSource extends RichSourceFunction<Student> {
//3、ParallelSourceFunction
//public class ZidingyiSource implements ParallelSourceFunction<Student> {
//4、RichParallelSourceFunction
//public class ZidingyiSource extends RichParallelSourceFunction<Student> {
// 推荐的
public class ZidingyiSource extends RichParallelSourceFunction<Student> {// ctrl + oprivate final Random random = new Random();private boolean flag = true;// 现在不用@Overridepublic void open(Configuration parameters) throws Exception {System.out.println("实现一些资源的开启");}// 现在不用@Overridepublic void close() throws Exception {System.out.println("实现一些资源的关闭");}@Overridepublic void run(SourceContext<Student> sourceContext) throws Exception {while (flag){String stu_id = UUID.randomUUID().toString();String stu_name = "Student_"+stu_id;int stu_age = random.nextInt(8)+10;long stu_timestamp = System.currentTimeMillis();Student student = new Student(stu_id,stu_name,stu_age,stu_timestamp);sourceContext.collect(student);Thread.sleep(1000);}}// 具体什么时候 会调用还不知道@Overridepublic void cancel() {flag = false;System.out.println("停止运行");}
}//调用
public class ZiDingYi {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// add + new DataStreamSource<Student> studentDataStreamSource = env.addSource(new ZidingyiSource());int parallelism = studentDataStreamSource.getParallelism();System.out.println(parallelism);// print之前与之后的并行度是不同的studentDataStreamSource.print().setParallelism(1);env.execute();}
}
cancel+open+close的调用时机
package com.bigdata.day02;import java.util.Objects;/*
* 1、这几个方法都会按照并行度调用多次 调度的次数 按照studentDataStreamSource的并行度
*
*/public class ZiDingYi {public static void main(String[] args) throws Exception {// 在上面案例的基础上实现StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<Student> studentDataStreamSource = env.addSource(new ZidingyiSource());// 此时就只会调用一次了studentDataStreamSource.setParallelism(1);// 此时打印也会有多个并行度(8个cpu)studentDataStreamSource.print();// 异步调用 此时会调用open方法JobExecutionResult execute = env.execute();JobClient flink_job = env.executeAsync("Flink Job");Thread.sleep(3000);// 此时会调用 cancel 和 close flink_job.cancel();}
}
kafkaSource
package com.bigdata.day02;import java.util.Properties;public class KafkaSource {public static void main(String[] args) throws Exception{//envStreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// properties Properties properties = new Properties();properties.setProperty("bootstrap.servers", "bigdata01:9092");properties.setProperty("group.id", "g1");// consumerFlinkKafkaConsumer<String> consumer= new FlinkKafkaConsumer<String>("yhedu",new SimpleStringSchema(),properties);// sourceDataStreamSource<String> dataStreamSource = env.addSource(consumer);dataStreamSource.filter(new FilterFunction<String>() {@Overridepublic boolean filter(String s) throws Exception {return s.contains("success");}}).print();env.execute();}
}
相关文章:
flink学习(6)——自定义source和kafka
概述 SourceFunction:非并行数据源(并行度只能1) --接口 RichSourceFunction:多功能非并行数据源(并行度只能1) --类 ParallelSourceFunction:并行数据源(并行度能够>1) --接口 RichParallelSourceFunction:多功能并行数据源(并行度能够>1) --类 【建议使用的】 ——…...
开发常见问题及解决
1.DBeaver 报Public Key Retrieval is not allowed 在使用DBeaver连接数据库时出现“Public Key Retrieval is not allowed”错误,主要是因为数据库连接配置的安全策略导致的。以下是详细的解释和解决方法: 错误原因 这个错误通常出现在连接MySQL数据…...
python excel接口自动化测试框架!
今天采用Excel继续写一个接口自动化测试框架。 设计流程图 这张图是我的excel接口测试框架的一些设计思路。 首先读取excel文件,得到测试信息,然后通过封装的requests方法,用unittest进行测试。 其中,接口关联的参数通过正则进…...
mybatis:You have an error in your SQL syntax;
完整报错You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near false, false, false, false, false, false, false, false, false, false, false, at line 1 SQL: INSERT INTO user …...
使用 Maven 开发 IntelliJ IDEA 插件
使用 Maven 开发 IntelliJ IDEA 插件的完整流程 1. 创建 Maven 项目 1.1 使用 IntelliJ 创建 Maven 项目 打开 IntelliJ IDEA,点击 File > New > Project。选择 Maven,填写项目名称和 GroupId,例如: GroupId: com.exampl…...
Windows修复SSL/TLS协议信息泄露漏洞(CVE-2016-2183) --亲测
漏洞说明: 打开链接:https://docs.microsoft.com/zh-cn/troubleshoot/windows-server/windows-security/restrict-cryptographic-algorithms-protocols-schannel 可以看到: 找到:应通过配置密码套件顺序来控制 TLS/SSL 密码 我们…...
uniapp生命周期:应用生命周期和页面生命周期
文章目录 1.应用的生命周期2.页面的生命周期 1.应用的生命周期 生命周期的概念:一个对象从创建、运行、销毁的整个过程被称为生命周期 生命周期函数:在生命周期中每个阶段会伴随着每一个函数的出发,这些函数被称为生命周期函数 所有页面都…...
基于SSM的婴幼儿用品商城系统+LW示例参考
1.项目介绍 功能模块:管理员(产品管理、产品分类、会员管理、订单管理、秒杀活动、文章管理、数据统计等)、普通用户(登录注册、个人中心、购物车、我的收藏、各类信息查看等)技术选型:SSM,jsp…...
【工具变量】城市供应链创新试点数据(2007-2023年)
一、测算方式:参考C刊《经济管理》沈坤荣和乔刚老师(2024)的做法,使用“供应链创新与应用试点”的政策虚拟变量(TreatPost)表征。若样本城市为试点城市,则赋值为 1,否则为 0…...
【carla生成车辆时遇到的问题】carla显示的坐标和carlaworld中提取的坐标y值相反
项目需要重新运行了一下generate_car.py的脚本,发现死活生成不了,研究了半天,发现脚本里面生成车辆的坐标值y和carla_ros_bridge_with_example_ego_vehicle.launch脚本打开的驾驶操控界面里面的y值正好是相反数! y1-y2 因为,我运行…...
Jira使用笔记二 ScriptRunner 验证问题创建角色
背景 最近在对公司Jira工作流改造,收到这么一个要求:某些问题类型只有某些角色可以创建。本来是想通过Jira内建的权限控制来处理的。结果点到权限页面,心都凉透了。 好吧,那只能上脚本了。最终使用ScriptRunner的Simple scripte…...
Java线程的使用
Java中的线程是用来实现多任务并发执行的机制。在Java中,主要有两种方式来创建和使用线程:实现Runnable接口和继承Thread类。 实现Runnable接口: 创建一个类,实现Runnable接口,并重写run()方法。在run()方法中定义线程…...
自动化测试工具Ranorex Studio(四十三)-RANOREXPATH编辑器5
代码示例 下面的代码示例将讲解如何使用Ranorex API来编写代码模块,或者是使用用户代码来扩展录制的模块。 在代码中使用对象库 使用对象库等待UI元素 建立Adapter来访问更多的属性和方法 为对象库元素建立一组Adapter 使用Validate类 强制一个测试用例失败 设置aut…...
超高流量多级缓存架构设计!
文章内容已经收录在《面试进阶之路》,从原理出发,直击面试难点,实现更高维度的降维打击! 文章目录 电商-多级缓存架构设计多级缓存架构介绍多级缓存请求流程负载均衡算法的选择轮询负载均衡一致性哈希负载均衡算法选择 应用层 Ngi…...
数据结构(Java)—— ArrayList
1.线性表 线性表( linear list)是n个具有相同特性的数据元素的有限序列。 线性表是一种在实际中广泛使用的数据结构,常见的线性表:顺序表、链表、栈、队列... 线性表在逻辑上是线性结构,也就说是连续的一条直线。但是在…...
实习冲刺第三十三天
102.二叉树的层序遍历 给你二叉树的根节点 root ,返回其节点值的 层序遍历 。 (即逐层地,从左到右访问所有节点)。 示例 1: 输入:root [3,9,20,null,null,15,7] 输出:[[3],[9,20],[15,7]]示例…...
Uniapp开发下拉刷新功能onPullDownRefresh/onReachBottom
文章目录 1.onPullDownRefresh2.onReachBottom 1.onPullDownRefresh 在 js 中定义 onPullDownRefresh 处理函数(和onLoad等生命周期函数同级),监听该页面用户下拉刷新事件。 需要在 pages.json 里,找到的当前页面的pages节点&am…...
什么是 C++ 中的函数对象?函数对象与普通函数有什么区别?如何定义和使用函数对象?
1) 什么是 C 中的函数对象?它有什么特点? 在 C 中,函数对象(也称为仿函数或 functor)是一种重载了 operator() 的对象。这意味着这些对象可以像函数一样被调用。函数对象通常用于需要传递行为(即代码&…...
PointNet++论文复现
✨✨ 欢迎大家来访Srlua的博文(づ ̄3 ̄)づ╭❤~✨✨ 🌟🌟 欢迎各位亲爱的读者,感谢你们抽出宝贵的时间来阅读我的文章。 我是Srlua小谢,在这里我会分享我的知识和经验。&am…...
【VUE】el-table表格内输入框或者其他控件规则校验实现
1、封装组件 1、规则校验一般基于form表单实现,因此需要给具体控件套一层form表单 新建组件input-required.vue,内容如下 <template><div><el-form ref"formRef" :model"form" :rules"formRules" label-…...
C1——优化3Dtiles透明度设置以实现管线可视化
1. 为什么需要调整3Dtiles透明度? 在地理信息系统(GIS)和三维可视化项目中,我们经常会遇到多层数据叠加显示的需求。比如在城市地下管线可视化场景中,地表建筑模型(3Dtiles)和地下管线网络需要同…...
Rainmeter皮肤模板循环控制:break/continue实现终极指南
Rainmeter皮肤模板循环控制:break/continue实现终极指南 【免费下载链接】rainmeter Desktop customization tool for Windows 项目地址: https://gitcode.com/gh_mirrors/ra/rainmeter Rainmeter作为一款强大的Windows桌面自定义工具,其皮肤模板…...
gte-base-zh场景应用:电商搜索与客服问答的语义匹配实战
gte-base-zh场景应用:电商搜索与客服问答的语义匹配实战 1. 电商场景中的语义匹配挑战 1.1 搜索不精准的痛点分析 在电商平台上,用户搜索"苹果手机"却看到水果苹果的图片,或者输入"轻薄笔记本"却返回游戏本࿰…...
告别拉伸变形!保姆级教程:为你的Unity Windows应用添加自定义窗口比例限制器
Unity Windows应用窗口比例锁定全攻略:从原理到避坑指南 你是否遇到过这样的尴尬场景——精心设计的UI在用户随意拉伸窗口后变得面目全非?作为Unity开发者,我们常常需要为Windows平台构建专业级应用,而窗口比例控制正是提升用户体…...
hot100——二分查找
4.寻找两个正序数组的中位数解题思路首先,题目中已经说明,是正序,那么nums1以及nums2中都是从小到大进行排列的;又因为题目中要求时间复杂度为O(log(mn)),一般看到这种时间复杂度是O(log……)形式的,基本上…...
FastAPI流式AI接口设计陷阱大全(2024高频真题+源码级调试实录)
第一章:FastAPI流式AI接口设计陷阱大全(2024高频真题源码级调试实录)流式响应被中间件静默截断 FastAPI 默认启用的 Starlette 中间件(如 HTTPSRedirectMiddleware 或自定义日志中间件)可能在未显式处理 StreamingResp…...
ViGEmBus终极指南:Windows游戏控制器虚拟化完整解决方案
ViGEmBus终极指南:Windows游戏控制器虚拟化完整解决方案 【免费下载链接】ViGEmBus Windows kernel-mode driver emulating well-known USB game controllers. 项目地址: https://gitcode.com/gh_mirrors/vi/ViGEmBus Windows游戏控制器虚拟化是许多PC游戏玩…...
APK Editor Studio:从入门到精通的完整Android应用编辑指南
APK Editor Studio:从入门到精通的完整Android应用编辑指南 【免费下载链接】apk-editor-studio Powerful yet easy to use APK editor for PC and Mac. 项目地址: https://gitcode.com/gh_mirrors/ap/apk-editor-studio 在Android应用开发和逆向工程领域&am…...
沈阳装修靠谱的机构
在沈阳装修新家,最怕遇到不靠谱的装修公司——工期拖延、增项不断、工艺粗糙、售后无门。想要省心、放心、安心地完成装修,选择一家经验丰富、工艺扎实、信誉良好的机构至关重要。在众多沈阳装修公司中,沈阳富田装饰装修工程有限公司以其深厚…...
OpenClaw多模型管理:Qwen3.5-4B-Claude与其他模型的协作方案
OpenClaw多模型管理:Qwen3.5-4B-Claude与其他模型的协作方案 1. 为什么需要多模型协作 去年冬天,当我第一次尝试用OpenClaw自动化处理技术文档时,发现单一模型很难兼顾所有任务场景。有些模型擅长代码生成但逻辑推理薄弱,有些长…...
