Flink DataSource介绍

介绍

Flink的Data Source（数据源、源算子）是Flink作业的起点，它定义了数据输入的来源。Flink可以从各种数据来源获取数据，例如文件系统、消息队列、数据库等。以下是对Flink Data Source的详细介绍：

概述：

  • Flink中的Data Source用于定义数据输入的来源。
  • 将数据源添加到Flink执行环境中，可以创建一个数据流。
  • Flink支持多种类型的数据源，包括内置数据源和自定义数据源。

内置数据源：

  • 基于集合构建：使用Flink的API（如fromCollection、fromElements等）将Java或Scala中的集合数据转化为数据流进行处理。
  • 基于文件构建：从文件系统中读取数据，支持多种文件格式，如CSV、JSON等。
  • 基于Socket构建：从Socket连接中读取数据，适用于实时流数据场景。

自定义数据源：

  • Flink允许用户通过实现SourceFunction接口或扩展RichParallelSourceFunction来自定义数据源。
  • 常见的自定义数据源包括从第三方系统连接器（如Kafka、RabbitMQ、MongoDB等）中读取数据。

添加数据源到Flink执行环境：

  • 使用StreamExecutionEnvironment.addSource(sourceFunction)方法将数据源添加到Flink执行环境中。
  • sourceFunction需要实现SourceFunction接口或扩展RichParallelSourceFunction。

数据流处理：

  • 一旦数据源被添加到Flink执行环境中，就可以创建一个数据流（DataStream）。
  • 接下来，可以使用Flink的各种算子（如map、filter、reduce等）对数据流进行转换处理。

输出结果：

  • 处理后的数据可以写入其他系统，如文件系统、数据库、消息队列等。
  • Flink支持多种输出方式，如使用DataStream的writeAsText、writeAsCsv等方法将数据写入文件，或使用Flink的连接器将数据写入Kafka、HBase等系统。

总之，Flink的Data Source是构建Flink数据流处理应用的重要组成部分。通过选择合适的数据源和输出方式，可以方便地构建高效、可靠的数据流处理应用。

样例

程序中添加数据源

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.addSource(new SourceMySQL()).print();env.execute("Flink add mysql sourc");

Flink 已经提供了若干实现好了的SourceFunctions也可以通过实现 SourceFunction 来自定义非并行的 source 或者实现 ParallelSourceFunction 接口或者扩展 RichParallelSourceFunction 来自定义并行的 source，

stream sources

StreamExecutionEnvironment 加载Source

基于集合：

  1. fromCollection(Collection)
  2. fromCollection(Iterator, Class)
  3. fromElements(T …)
  4. fromParallelCollection(SplittableIterator, Class)
  5. generateSequence(from, to)
    例如：
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<Event> input = env.fromElements(new Event(1, "ba", 4.0),new Event(2, "st", 5.0),new Event(3, "fo", 6.0),...
);

文件

  1. readTextFile(String filePath)
  2. readTextFile(String filePath, String charsetName)
  3. readFile(FileInputFormat inputFormat, String filePath)
    样例：
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<String> text = env.readTextFile("file:///path/file");

Socket

  1. socketTextStream(String hostname, int port)
  2. socketTextStream(String hostname, int port, String delimiter)
  3. socketTextStream(String hostname, int port, String delimiter, long maxRetry)
    样例：
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStream<Tuple2<String, Integer>> dataStream = env.socketTextStream("localhost", 8888) // 监听 localhost 的 8888端口过来的数据.flatMap(new Splitter()).keyBy(0).timeWindow(Time.seconds(5)).sum(1);

自定义资源

通过实现 SourceFunction 来自定义非并行的 source 或者实现 ParallelSourceFunction 接口或者扩展 RichParallelSourceFunction 来自定义并行的 source。
继承类

SourceFunction 非并行

SourceFunction 是一个用于定义数据源的函数接口。这个接口的实现通常负责从外部系统（如 Kafka、文件系统、数据库等）读取数据，并将这些数据作为 Flink 流处理或批处理作业的输入。

SourceFunction 通常与 Flink 的 DataStream API 一起使用，以定义和构建数据流处理任务。尽管 Flink 的内部实现和 API 可能会随着版本的更新而有所变化，但通常，一个 SourceFunction 会被实现为：

  1. 创建一个可以持续生成数据或等待外部数据到达的组件。
  2. 当有新数据到达时，将数据作为 Flink 的 DataStream 的一部分进行发射（emit）。

在 Flink 的内部，SourceFunction 可能会有多种不同的实现，具体取决于其要处理的数据源类型。例如，对于 Kafka 这样的消息队列，Flink 提供了专门的 Kafka 连接器和相应的 SourceFunction 实现，用于从 Kafka 主题中读取数据。

在实现自定义的 SourceFunction 时，你需要考虑以下几个方面：

  • 数据源的连接和断开连接逻辑。
  • 数据的读取和解析逻辑。
  • 如何在 Flink 运行时环境中优雅地处理可能的错误和失败。
  • 如何将读取的数据转换为 Flink 可以理解的格式（如 Tuple、POJO 或其他自定义类型）。

ParallelSourceFunction 并行

ParallelSourceFunction 是一个接口，用于定义并行数据源的行为。这个接口允许你创建自定义的数据源，这些数据源能够并行地读取数据并传递给 Flink 的数据处理管道。

ParallelSourceFunction 继承自 SourceFunction，但增加了并行处理的能力。当 Flink 任务需要并行处理多个数据流时，你可以通过实现 ParallelSourceFunction 来创建并行数据源。

Flink 还提供了一个 RichParallelSourceFunction 抽象类，它是 ParallelSourceFunction 的子类，并提供了更多的生命周期方法和上下文信息。使用 RichParallelSourceFunction 可以让你更容易地管理你的并行数据源，因为它提供了诸如 open()、close() 和 cancel() 等方法，这些方法可以在数据源的生命周期中的不同阶段被调用。

下面是一个简单的示例，演示了如何使用 RichParallelSourceFunction 创建一个并行数据源，该数据源生成递增的数字：

import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;  
public class IncrementingNumberSource extends RichParallelSourceFunction<Long> {  private volatile boolean running = true;  private long count = 0L;  @Override  public void run(SourceContext<Long> ctx) throws Exception {  while (running) {  synchronized (ctx.getCheckpointLock()) {  ctx.collect(count++);  // 这里可以添加一些休眠或其他逻辑来控制数据的生成速度  Thread.sleep(100);  }  }  }  @Override  public void cancel() {  running = false;  }  
}

在这个示例中，IncrementingNumberSource 类继承了 RichParallelSourceFunction，并覆盖了 run() 和 cancel() 方法。在 run() 方法中，我们创建了一个无限循环来生成递增的数字，并使用 ctx.collect() 方法将每个数字发送到 Flink 的数据处理管道中。在 cancel() 方法中，我们设置了一个标志来停止 run() 方法中的循环，以便在需要时可以优雅地关闭数据源。

自定义资源DEMO

/*** Desc: 自定义 source mysql 数据*/
public class SourceMySQL extends RichSourceFunction<Map<String, Object>> {PreparedStatement ps;private Connection connection;/*** open 方法中建立连接，这样不用每次 invoke 的时候都要建立连接和释放连接。*/@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);connection = MySQLUtil.getConnection("com.mysql.jdbc.Driver","jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8","root","123456");String sql = "select * from ST;";ps = this.connection.prepareStatement(sql);}/*** 关闭连接和释放资源的动作*/@Overridepublic void close() throws Exception {super.close();if (connection != null) {connection.close();}if (ps != null) {ps.close();}}/*** DataStream 从run方法用来获取数据*/@Overridepublic void run(SourceContext<Map<String, Object>> ctx) throws Exception {ResultSet resultSet = ps.executeQuery();while (resultSet.next()) {Map<String, Object> rs = new HashMap<>();rs.put("id", resultSet.getInt("id"));rs.put("name", resultSet.getString("name").trim());ctx.collect(rs);}}@Overridepublic void cancel() {}
}

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

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

相关文章

YTM32的片内flash应用答疑 - 释疑efm_sts[accerr]寄存器位

YTM32的片内flash应用答疑 - 释疑efm_sts[accerr]寄存器位

YTM32的片内flash应用答疑 - 释疑efm_sts[accerr]寄存器位 文章目录 YTM32的片内flash应用答疑 - 释疑efm_sts[accerr]寄存器位IntroductionConceptConclusion Introduction 之前有客户在基于ytm32b1le05微控制器做ota方案&#xff0c;其中在擦写片内flash模块时&#xff0c;需…
阅读更多...
SpringCloudAlibaba:4.2云原生网关higress的基本使用

SpringCloudAlibaba:4.2云原生网关higress的基本使用

概述 简介 Higress是基于阿里内部的Envoy Gateway实践沉淀、以开源Istio Envoy为核心构建的下一代云原生网关&#xff0c; 实现了流量网关 微服务网关 安全网关三合一的高集成能力&#xff0c;深度集成Dubbo、Nacos、Sentinel等微服务技术栈 定位 在虚拟化时期的微服务架构…
阅读更多...
138.随机链表的复制

138.随机链表的复制

/*** Definition for a Node.* struct Node {* int val;* struct Node *next;* struct Node *random;* };*/ typedef struct Node Node; struct Node* copyRandomList(struct Node* head) {Node* curhead;//拷贝节点插入到原节点后面while(cur){Node* copy(Node*)m…
阅读更多...
鹤壁哪里有在线教育系统开发公司,开办培训机构需要准备哪些工作？

鹤壁哪里有在线教育系统开发公司,开办培训机构需要准备哪些工作？

很多开办培训机构的朋友&#xff0c;前期并没有做充足的准备&#xff0c;只是觉得&#xff1a;这个小区不错&#xff0c;是高收入群体&#xff0c;孩子多&#xff0c;可以试试;我是老师&#xff0c;开培训机构有资源&#xff0c;可以试试;这边有一家培训机构做的不错&#xff0…
阅读更多...
Apache正向代理是什麼？

Apache正向代理是什麼？

Apache正向代理&#xff08;Forward Proxy&#xff09;是一種位於客戶端和原始伺服器之間的伺服器&#xff0c;通常用於客戶端和互聯網之間的連接&#xff0c;幫助內部網路的用戶訪問互聯網&#xff0c;例如家庭網路或公司網路&#xff0c;幫助用戶進行安全、有效的流覽。 Apa…
阅读更多...
基于参数化建模的3D产品组态实现

基于参数化建模的3D产品组态实现

我们最近为荷兰设计师家具制造商 KILO 发布了基于网络的 3D 配置器的第一个生产版本。我们使用了 Salsita 3D 配置器&#xff0c;这是一个内部 SDK&#xff0c;使新的 3D 配置器的实施变得轻而易举。虽然它给我们带来了巨大帮助&#xff0c;但我们仍然面临一些有趣的挑战。 NSD…
阅读更多...
C语言【文件操作】（1）

C语言【文件操作】（1）

文章目录 1.为什么使用文件2.文件是什么&#xff1f;2.1程序文件2.2数据文件 3.二进制文件和文本文件4.文件的打开和关闭4.1流和标准流流标准流 4.2文件指针4.3文件的打开和关闭 结语 1.为什么使用文件 很简单 长久的存储数据 如果没有文件&#xff0c;我们写程序所产生的数据…
阅读更多...
企业内部适用的五大知识库工具测评推荐

企业内部适用的五大知识库工具测评推荐

随着企业规模的不断扩大和业务复杂性的增加&#xff0c;要想更高效地进行企业管理就不得不使用知识库管理工具。本文将对五款企业内部适用的知识库工具进行测评推荐&#xff0c;帮助企业选择出更适合自己的知识库管理工具。 一、Helplook AI知识库 Helplook AI知识库是一款搭建…
阅读更多...
线程创建、join、detach

线程创建、join、detach

线程的创建 进程的生命周期是和主线程相同的。进程的开启伴随着主线程的出现&#xff0c;主线程执行完毕代表着进程的结束。在编译器上&#xff0c;主线程是从main函数开始&#xff0c;当主线程从main函数上返回&#xff0c;就代表着进程的结束。一般情况下&#xff0c;如果主…
阅读更多...
4种常⽤Java线程锁的特点，

4种常⽤Java线程锁的特点，

1 提供了多种多线程锁机制的实现⽅式 synchronizedReentrantLockSemaphoreAtomicInteger
阅读更多...
【LeetCode刷题记录】124. 二叉树中的最大路径和

【LeetCode刷题记录】124. 二叉树中的最大路径和

124 二叉树中的最大路径和 二叉树中的 路径 被定义为一条节点序列&#xff0c;序列中每对相邻节点之间都存在一条边。同一个节点在一条路径序列中 至多出现一次 。该路径 至少包含一个 节点&#xff0c;且不一定经过根节点。 路径和 是路径中各节点值的总和。 给你一个二叉树的…
阅读更多...
【Java开发的我出书啦，各位同仁快过来围观】！！！

【Java开发的我出书啦，各位同仁快过来围观】！！！

文章目录 &#x1f50a;博主介绍&#x1f964;本文内容出书的目的出书的过程书籍的内容 &#x1f4e5;博主的话 &#x1f50a;博主介绍 文章目录 &#x1f50a;博主介绍&#x1f964;本文内容出书的目的出书的过程书籍的内容 &#x1f4e5;博主的话 &#x1f33e;阅读前&#x…
阅读更多...
linux系统的备份管理

linux系统的备份管理

centos7系统备份与还原 1、系统备份 下面是我用来备份系统的完整命令&#xff1a; # cd /# tar -zcvpf centos7.tgz --exclude/proc --exclude/lostfound --exclude/mnt --exclude/sys --excludecentos7.tgz / “tar”当然就是我们备份系统所使用的程序了。 “cvpfz”是tar的选…
阅读更多...
【C++】-QT串口通信-008

【C++】-QT串口通信-008

1
阅读更多...
「网络流 24 题」太空飞行计划 【最大权值闭合图】

「网络流 24 题」太空飞行计划 【最大权值闭合图】

「网络流 24 题」太空飞行计划 题意 有 n n n 个实验 和 m m m 个器械&#xff0c;每个实验都需要若干个指定的器械才能进行 实验 i i i 的盈利为 p i p_i pi​&#xff0c; 器械 j j j 的花销为 c j c_j cj​ 找出纯利润最大的实验计划 思路 这是非常典型的最大权值…
阅读更多...
如何定时关闭程序

如何定时关闭程序

首先&#xff0c;需要用到的这个工具&#xff1a; 度娘网盘 提取码&#xff1a;qwu2 蓝奏云 提取码&#xff1a;2r1z 前面的流程步骤参考这条攻略&#xff1a; 快捷自由定时重启、注销、关机 只不过最后的地方&#xff0c;选择 关闭程序 &#xff0c;再填写程序名称即可 补…
阅读更多...
2024-03-23青少年软件编程（Python语言）等考（四级）解析

2024-03-23青少年软件编程（Python语言）等考（四级）解析

2024-03-23青少年软件编程（Python语言）等考（四级）解析一、单选题(共25题，共50分) 1.运行如下代码，若输入整数3，则最终输出的结果为？（ C ） def f(x):if x==1:s=1else:s=f(x-1)*xreturn s n=int(input("请输入一个大于1的整数：")) print(f(n)+f(n-1)) A.2…
阅读更多...
全栈开发之路——前端篇（5）组件间通讯和接口等知识补充

全栈开发之路——前端篇（5）组件间通讯和接口等知识补充

全栈开发一条龙——前端篇 第一篇&#xff1a;框架确定、ide设置与项目创建 第二篇&#xff1a;介绍项目文件意义、组件结构与导入以及setup的引入。 第三篇&#xff1a;setup语法&#xff0c;设置响应式数据。 第四篇&#xff1a;数据绑定、计算属性和watch监视 辅助文档&…
阅读更多...
Linux中的软连接和硬链接

Linux中的软连接和硬链接

一、软和硬链接连接 在Linux系统中&#xff0c;软连接&#xff08;符号链接&#xff09;和硬链接是文件系统中两种不同类型的链接&#xff0c;它们用于创建对文件的引用。下面详细解释这两种链接的特点和区别&#xff1a;、 软连接&#xff08;符号链接&#xff09; 定义&…
阅读更多...
项目中遇到的问题

项目中遇到的问题

web项目中请求线程到service层的时候远程调用服务之前是串行化执行每个任务都要get阻塞等待任务完成&#xff0c;举例当用户在购物车页面点击去结算就会请求后台toTrade请求获取订单确认的详情数据并渲染到订单详情页&#xff0c;现在在toTrade请求中使用异步任务编排Completab…
阅读更多...
最新文章

Copyright @ 2022~2023