当前位置: 首页 > news >正文

1. Flink自定义Source

一. Source 简介

DataStream是Flink的低级API,用于进行数据的实时处理,Flink编程模型分为Source、Transformation、Sink三个部分,如下图所示。
在这里插入图片描述

默认Flink提供了大量的内置Source,常见的Source如下:

  • 基于文件的Source
  • 基于Socket的Source
  • 基于集合的Source
  • 基于Kafka消息队列的Source

当以上内置Source不能满足业务需要时,可以实现自定义Source。

Flink中有关Source的接口类的继承关系如下:
在这里插入图片描述

  • SourceFunction:单并行度Source的基类
  • RichSourceFunction:单并行度增强型Source的基类
  • ParallelSourceFunction:多并行度Source的基类
  • RichParallelSourceFunction:多并行度增强型Source的基类
二. 自定义单并行度Source

自定义单并行度的source需要实现SourceFunction接口。

代码实现

MySource.java

package flink.basic.source;import org.apache.flink.streaming.api.functions.source.SourceFunction;
import java.util.Random;public class MySource implements SourceFunction<String> {boolean running = true;@Overridepublic void run(SourceContext<String> ctx) throws Exception {Random random = new Random();while (running) {// "Num"加上0~100的随机数生成一个字符串ctx.collect("Num: " + random.nextInt(100));Thread.sleep(1000);}}@Overridepublic void cancel() {running = false;}
}

SourceDemo.java

package flink.basic.source;import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class SourceDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env= StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> source = env.addSource(new MySource());source.print();env.execute("source_demo");}
}

运行结果

5> Num: 62
6> Num: 91
7> Num: 13
8> Num: 53
三. 自定义多并行度Source

自定义多并行度的source需要实现ParallelSourceFunction接口。

代码实现

MyParallelSource.java

package flink.basic.source;import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import java.util.Random;public class MyParallelSource implements ParallelSourceFunction<String> {boolean running = true;@Overridepublic void run(SourceContext<String> ctx) throws Exception {Random random = new Random();while (running) {ctx.collect("Num: " + random.nextInt(100));Thread.sleep(1000);}}@Overridepublic void cancel() {running = false;}
}

SourceDemo.java

package flink.basic.source;import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class SourceDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env= StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> source = env.addSource(new MyParallelSource());source.print();env.execute("source_demo");}
}

运行结果

7> Num: 43
8> Num: 30
1> Num: 92
2> Num: 50
5> Num: 39
6> Num: 6
4> Num: 20
3> Num: 2
四. 自定义单并行度增强型Source

增强型Source额外提供了open和close方法,可以用于自定义Source的初始化和清理工作。单并行度增强型Source需要实现RichSourceFunction接口。下面演示实现读取mysql表的单并行度Source。

在mysql中创建student表,并插入三条数据。

create table student (id int primary key,name varchar(50),age int
);insert into student values(1, "name1", 20),(2, "name2", 30), (3, "name3", 15);

实现代码

Student.java

package flink.basic.source;public class Student {private int id;private String name;private int age;public Student(int id, String name, int age) {this.id = id;this.name = name;this.age = age;}public Student() {}public int getId() {return id;}public void setId(int id) {this.id = id;}public String getName() {return name;}public void setName(String name) {this.name = name;}public int getAge() {return age;}public void setAge(int age) {this.age = age;}@Overridepublic String toString() {return "Student{" +"id=" + id +", name='" + name + '\'' +", age=" + age +'}';}
}

MysqlSource.java

package flink.basic.source;import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;public class MysqlSource extends RichSourceFunction<Student> {Connection conn;Statement stmt;@Overridepublic void open(Configuration parameters) throws Exception {Class.forName("com.mysql.cj.jdbc.Driver");String url = "jdbc:mysql://192.168.47.130:3306/test";String user = "root";String password = "root";conn = DriverManager.getConnection(url,user,password);stmt = conn.createStatement();}@Overridepublic void run(SourceContext<Student> ctx) throws Exception {ResultSet rs = stmt.executeQuery("select * from student");while (rs.next()) {int id = rs.getInt("id");String name = rs.getString("name");int age = rs.getInt("age");ctx.collect(new Student(id, name, age));}rs.close();}@Overridepublic void cancel() {}@Overridepublic void close() throws Exception {stmt.close();conn.close();}
}

SourceDemo.java

package flink.basic.source;import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class SourceDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env= StreamExecutionEnvironment.getExecutionEnvironment();// 添加mysql SourceDataStreamSource<Student> source = env.addSource(new MysqlSource());source.print();env.execute("source_demo");}
}

运行结果

1> Student{id=3, name='name3', age=15}
8> Student{id=2, name='name2', age=30}
7> Student{id=1, name='name1', age=20}

相关文章:

1. Flink自定义Source

一. Source 简介 DataStream是Flink的低级API&#xff0c;用于进行数据的实时处理&#xff0c;Flink编程模型分为Source、Transformation、Sink三个部分&#xff0c;如下图所示。 默认Flink提供了大量的内置Source&#xff0c;常见的Source如下&#xff1a; 基于文件的Sour…...

关于LinuxWindows双系统在八月更新后出现的问题

问题描述类似于&#xff1a;Verifying shim SBAT data failed: If you are, this is caused by a reported problem in the August update if you can get into Windows, either uninstall the August update, or open Command Prompt as administrator and run this command,…...

VMware:如何在CentOS7上开启22端口

打开虚拟机&#xff1a;【编辑】【虚拟机网络设置】 其中填入的虚拟机IP地址是虚拟机中centos的IP地址&#xff0c;虚拟机端口为需要映射的centos端口 配置好之后保存&#xff0c;打开宿主机 win cmd telnet 192.168.1.26 22 如果出现上述窗口&#xff0c;则说明已经成功开放…...

ubuntu远程桌面开启opengl渲染权限

背景 最近用windows的【远程桌面连接】登录ubuntu后&#xff08;xrdp协议&#xff09;&#xff0c;发现gl环境是集显的&#xff0c;但是本地登录ubuntu桌面后是独显&#xff08;英伟达&#xff09;&#xff0c;想要在远程桌面上也用独显渲染环境。 一、查看是独显还是集显环境…...

从小学题到技术选型哲学:以智能客服系统为例,解读相关AI技术栈20241211

&#x1f9e0;&#x1f4a1;从小学题到技术选型哲学&#xff1a;以智能客服系统为例&#xff0c;解读相关AI技术栈 引言&#xff1a;从小学数学题到技术智慧 &#x1f4da;✨ 在小学数学题中&#xff0c;有这样一道问题&#xff1a; “一个长方形变成平行四边形后&#xff0c…...

【C语言练习(5)—回文数判断】

C语言练习&#xff08;5&#xff09; 文章目录 C语言练习&#xff08;5&#xff09;前言问题问题解析结果总结 前言 通过回文数练习&#xff0c;巩固数字取余和取商如何写代码 问题 输入一个五位数判断是否为回文数&#xff1f; 问题解析 回文数是指正读反读都一样的整数。…...

【Rust 学习笔记】Rust 基础数据类型介绍——数组、向量和切片

博主未授权任何人或组织机构转载博主任何原创文章&#xff0c;感谢各位对原创的支持&#xff01; 博主链接 博客内容主要围绕&#xff1a; 5G/6G协议讲解 高级C语言讲解 Rust语言讲解 文章目录 Rust 基础数据类型介绍——数组、向量和切片一、数组、向量和…...

2024年特别报告,「十大生活方式」研究数据报告

“一朵花成轻奢品、一只玩偶掀抢购狂潮、一片荒地变文旅圣地…” 近年爆火的野兽派、Jellycat、阿那亚等诸多品牌&#xff0c;与消费者选择的生活方式息息相关。 今年小红书的内容种草、直播电商&#xff0c;也都依循着“生活方式”的轨迹。生活方式的价值所向&#xff0c;可…...

R中单细胞RNA-seq分析教程 (5)

引言 本系列开启R中单细胞RNA-seq数据分析教程[1]&#xff0c;持续更新&#xff0c;欢迎关注&#xff0c;转发&#xff01; 10. 伪时间细胞排序 如前所述&#xff0c;在 UMAP 嵌入中看到的背侧端脑细胞形成的类似轨迹的结构&#xff0c;很可能代表了背侧端脑兴奋性神经元的分化…...

openpnp - Too many misdetects - retry and verify fiducial/nozzle tip detection

文章目录 openpnp - Too many misdetects - retry and verify fiducial/nozzle tip detection概述笔记环境光最好弱一些在设备标定时&#xff0c;吸嘴上不要装绿色屏蔽片如果吸嘴不在底部相机中间&#xff0c;先检查设置底部相机坐标调整底部相机坐标 吸嘴校验的细节底部相机坐…...

不与最大数相同的数字之和

不与最大数相同的数字之和 C语言代码C 语言代码Java语言代码Python语言代码 &#x1f490;The Begin&#x1f490;点点关注&#xff0c;收藏不迷路&#x1f490; 输出一个整数数列中不与最大数相同的数字之和。 输入 输入分为两行&#xff1a; 第一行为N(N为接下来数的个数&…...

CSS学习记录11

CSS布局 - display属性 display属性是用于控制布局的最终要的CSS属性。display 属性规定是否/如何显示元素。每个HTML元素都有一个默认的display值&#xff0c;具体取决于它的元素类型。大多数元素的默认display值为block 或 inline。 块级元素&#xff08;block element&…...

D95【python 接口自动化学习】- pytest进阶之fixture用法

day95 pytest的fixture详解&#xff08;二&#xff09; 学习日期&#xff1a;20241210 学习目标&#xff1a;pytest基础用法 -- pytest的fixture详解&#xff08;二&#xff09; 学习笔记&#xff1a; fixture(autouseTrue) func的autouse是TRUE时&#xff0c;所有函数方法…...

Abaqus断层扫描三维重建插件CT2Model 3D V1.1版本更新

更新说明 Abaqus AbyssFish CT2Model3D V1.1版本更新新增对TIF、TIFF图像文件格式的支持。本插件用户可免费获取升级服务。 插件介绍 插件说明&#xff1a; Abaqus基于CT断层扫描的三维重建插件CT2Model 3D 应用案例&#xff1a; ABAQUS基于CT断层扫描的细观混凝土三维重建…...

隐式对象和泛型

implicit object 作用&#xff1a; case class DatabaseConfig(driver:String,url:String)//作为函数的隐士参数的默认值implicit object MySqlDefault extends DatabaseConfig("mysql","localhost:443")def getConn(implicit config: DatabaseConfig):Uni…...

CSS的颜色表示方式

以下介绍几种常见的CSS颜色表示方式&#xff1a; 颜色名称 html和css规范中定义了147种可用的颜色名用的相对较少 16进制表示 css三原色&#xff1a;红、绿、蓝16进制的颜色值&#xff1a; #rrggbb16进制整数规定颜色成分&#xff0c;所有的值均介于 00 - ff 之间&#xff…...

单链表常见面试题 —— LeetCode

一.删除链表中与val相等的所有节点 1.题目描述 ----- 203. 移除链表元素 - 力扣&#xff08;LeetCode&#xff09; 给你一个链表的头节点 head 和一个整数 val &#xff0c;请你删除链表中所有满足 Node.val val 的节点&#xff0c;并返回 新的头节点 。 列表中的节点数目在范…...

Pydantic中的discriminator:优雅地处理联合类型详解

Pydantic中的discriminator&#xff1a;优雅地处理联合类型详解 引言1. 什么是discriminator&#xff1f;2. 基本使用示例3. discriminator的工作原理4. 更复杂的实际应用场景5. 使用建议6. 潜在陷阱和注意事项结论最佳实践 引言 在Python的类型系统中&#xff0c;有时我们需要…...

pgloader SQLSERVER -> PostgreSQL 配置文件样例

pgloader 是什么&#xff1f;安装和基本用户法可以去其他同道的blog上去看&#xff0c;这里不占用网络空间了。刚开始用官方的文档读起还是很费劲的&#xff0c;所以把常用的配置例子放在这里。 官方文档&#xff1a;https://pgloader.readthedocs.io/en/latest/index.html 迁…...

APP、小程序对接聚合广告平台,有哪些广告变现策略?

开发者对接聚合广告平台&#xff0c;可以让自身流量价值最大化&#xff0c;获得更多的广告曝光机会&#xff0c;对接单一的广告联盟容易造成广告填充不足&#xff0c;收益不稳定的问题。#APP广告变现# APP开发者根据应用的生命周期、用户特征和产品定位&#xff0c;选择最适合…...

League Akari:英雄联盟玩家的智能效率工具集,从自动秒选到战绩分析的全能助手

League Akari&#xff1a;英雄联盟玩家的智能效率工具集&#xff0c;从自动秒选到战绩分析的全能助手 【免费下载链接】League-Toolkit 兴趣使然的、简单易用的英雄联盟工具集。支持战绩查询、自动秒选等功能。基于 LCU API。 项目地址: https://gitcode.com/gh_mirrors/le/L…...

VCAM虚拟摄像头:3大创新功能解锁安卓摄像头的无限应用场景

VCAM虚拟摄像头&#xff1a;3大创新功能解锁安卓摄像头的无限应用场景 【免费下载链接】com.example.vcam 虚拟摄像头 virtual camera 项目地址: https://gitcode.com/gh_mirrors/co/com.example.vcam VCAM虚拟摄像头是一款基于Xposed框架的安卓虚拟相机解决方案&#x…...

Unity Enter Play Mode Settings 搭配手动Reload全攻略:既保速度又保数据安全

Unity开发效率革命&#xff1a;Enter Play Mode Settings与智能Reload的黄金组合 在Unity项目开发的中后期&#xff0c;随着代码量膨胀和资源规模增长&#xff0c;每次按下Play按钮后的等待时间逐渐成为效率杀手。传统工作流中&#xff0c;脚本修改后的自动Reload机制像一把双刃…...

2025新算法TOC优化VMD实战:六种熵值评估信号分解,一键Matlab出图

1. 为什么需要优化VMD参数&#xff1f; 第一次接触VMD&#xff08;Variational Mode Decomposition&#xff09;时&#xff0c;我和很多初学者一样被它的参数调优问题困扰。记得当时处理一组轴承振动信号&#xff0c;手动试了十几组K值和α值&#xff0c;结果要么模态分解不彻底…...

springboot基于Vue的大学生心理健康交流系统的设计与实现

目录系统架构设计数据库设计核心功能模块前端界面开发测试部署方案项目进度规划注意事项项目技术支持源码获取详细视频演示 &#xff1a;文章底部获取博主联系方式&#xff01;同行可合作系统架构设计 采用前后端分离架构&#xff0c;后端使用SpringBoot框架提供RESTful API接…...

ConcurrentHashMap讲解

在 Java 并发编程中&#xff0c;ConcurrentHashMap 是高频使用的线程安全 Map 实现&#xff0c;也是面试中几乎必问的核心知识点。它完美解决了 HashMap 线程不安全、Hashtable 性能极差的痛点&#xff0c;在高并发场景下实现了安全与性能的平衡。本文将从设计背景、JDK1.7/JDK…...

Grok-1深度实战指南:3140亿参数混合专家模型的高级部署与优化

Grok-1深度实战指南&#xff1a;3140亿参数混合专家模型的高级部署与优化 【免费下载链接】grok-1 马斯克旗下xAI组织开源的Grok AI项目的代码仓库镜像&#xff0c;此次开源的Grok-1是一个3140亿参数的混合专家模型 项目地址: https://gitcode.com/GitHub_Trending/gr/grok-1…...

别被TMOS吓到!拆解沁恒CH579蓝牙例程,看事件驱动如何简化你的代码

别被TMOS吓到&#xff01;拆解沁恒CH579蓝牙例程&#xff0c;看事件驱动如何简化你的代码 第一次打开沁恒CH579的蓝牙例程&#xff0c;看到满屏的TMOS_前缀函数和eventID定义&#xff0c;是不是瞬间头皮发麻&#xff1f;作为从51单片机转战蓝牙开发的工程师&#xff0c;我完全理…...

如何为你的单片机项目选择最佳通信协议?I²C、SPI、UART全解析

单片机通信协议深度指南&#xff1a;从理论到实战的精准选择策略 当你的单片机需要与外部世界对话时&#xff0c;选择正确的通信协议就像为不同场合挑选合适的语言——商务会议需要正式严谨&#xff0c;朋友聊天则讲究轻松随意。在嵌入式系统设计中&#xff0c;UART、IC和SPI这…...

从CTF逆向实战出发:手把手教你用Python脚本破解RC4和Base58加密(附完整代码)

从CTF逆向实战出发&#xff1a;手把手教你用Python脚本破解RC4和Base58加密&#xff08;附完整代码&#xff09; 在CTF竞赛中&#xff0c;逆向工程题目往往涉及各种加密算法的识别与破解。本文将聚焦两种常见加密方式——RC4和Base58&#xff0c;通过Python脚本实现从算法识别到…...