【天衍系列 01】深入理解Flink的 FileSource 组件:实现大规模数据文件处理

文章目录

  • 01 基本概念
  • 02 工作原理
  • 03 数据流实现
  • 04 项目实战
    • 4.1 项目结构
    • 4.2 maven依赖
    • 4.3 StreamFormat读取文件数据
    • 4.4 BulkFormat读取文件数据
    • 4.5 使用小结
  • 05 数据源比较
  • 06 总结

01 基本概念

Apache Flink 是一个流式处理框架,被广泛应用于大数据领域的实时数据处理和分析任务中。在 Flink 中,FileSource 是一个重要的组件,用于从文件系统中读取数据并将其转换为 Flink 的数据流。本文将深入探讨 FileSource 的工作原理、用法以及与其他数据源的比较。

02 工作原理

FileSource 是 Flink 提供的一种用于从文件系统中读取数据的源。它能够处理各种类型的文件,包括文本文件、压缩文件、序列文件等。FileSource 的工作原理可以概括为以下几个步骤:

1.文件分配(File Assignment)

在 Flink 集群中,每个任务都会负责读取文件的一个分片。FileSource 会根据文件的大小和数量将文件分配给不同的任务进行处理。

2.并行读取(Parallel Reading)

每个任务会并行地读取分配给它的文件分片。这意味着文件中的数据会被同时读取,从而提高了整体的读取速度和处理效率。

3.数据解析(Data Parsing)

读取的数据会经过解析器进行解析,将其转换为 Flink 中的数据结构,如 DataSet 或 DataStream。

4.数据分发(Data Distribution)

解析后的数据会被分发到后续的算子中进行进一步的处理和分析。

03 数据流实现

  • 有界流(Bounded Streams)

    有界流是指具有明确结束点的数据流,即数据流在某个时刻会结束,数据量是有限的。例如,从静态文件、数据库或有限数据集中读取的数据流就是有界流。有界流的特点包括:

    • 数据量是有限的,流的结束点是已知的。
    • 可以对整个数据流进行批处理式的分析和处理,因为所有数据都可用且有限。
    • 可以使用批处理算法和优化技术,例如排序、分组聚合等。
  • 无界流(Unbounded Streams)

    无界流是指没有明确结束点的数据流,即数据流会持续不断地产生,数据量可能是无限的。例如,实时传感器数据、日志流、消息队列中的数据等都是无界流。无界流的特点包括:

    • 数据源持续不断地产生数据,流没有明确的结束点。
    • 通常用于实时流式处理,要求系统能够实时处理数据并在流中进行持续的分析和计算。
    • 需要采用流式处理的技术和算法,例如窗口计算、流式聚合、事件时间处理等。
  • 不同数据流实现

    • 创建一个 File Source 时, 默认情况下,Source 为有界/批的模式;

      //创建一个FileSource数据源,并设置为批模式,读取完文件后结束
      final FileSource<String> source = FileSource.forRecordStreamFormat(...).build();
      
    • 设置参数monitorContinuously(Duration.ofMillis(5)) 可以把 Source 设置为持续的流模式

      //创建一个FileSource数据源,并设置为流模式,每隔5分钟检查路径新文件,并读取
      final FileSource<String> source = FileSource.forRecordStreamFormat(...).monitorContinuously(Duration.ofMillis(5))  .build();   
      

04 项目实战

1.FileSource支持多种数据格式数据读取与解析,本期以Text File文件为例展开。
2.jdk版本11
3.Flink版本1.18.0
4.下面是两个简单的示例代码,演示如何在 Flink 中使用 FileSource 读取文件数据

4.1 项目结构

在这里插入图片描述

4.2 maven依赖

<!-- flink读取Text File文件依赖 start-->
<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-files</artifactId><version>1.18.0</version>
</dependency>
<!-- flink读取Text File文件依赖 end--><!-- flink基础依赖 start -->
<dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>1.18.0</version>
</dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-scala_2.12</artifactId><version>1.18.0</version>
</dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients</artifactId><version>1.18.0</version>
</dependency><!-- flink基础依赖 end -->

4.3 StreamFormat读取文件数据

  • StreamFormat从文件流中读取文件内容。它是最简单的格式实现, 并且提供了许多拆箱即用的特性(如 Checkpoint 逻辑),但是限制了可应用的优化(例如对象重用,批处理等等)。
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.time.Duration;/*** 描述:* flink集成FileSource & forRecordStreamFormat使用 & 流模式* StreamFormat:从文件流中读取文件内容。它是最简单的格式实现,* 并且提供了许多拆箱即用的特性(如 Checkpoint 逻辑),* 但是限制了可应用的优化(例如对象重用,批处理等等)。** @author 浅夏的猫* @version 1.0.0* @date 2024-02-07 15:30:22*/
public class FileSourceRecordStreamingJob {public static void main(String[] args) throws Exception {// 创建 需要读取的文件路径PathPath path = new Path("D:\\flink\\file_source.txt");// 创建 读取文件的格式函数TextLineInputFormat textLineInputFormat = new TextLineInputFormat();// 创建 FileSourceFileSource<String> fileSource = FileSource.forRecordStreamFormat(textLineInputFormat, path)//放开注释则使用流模式,每隔5分钟检查是否有新文件否则默认使用批模式
//                .monitorContinuously(Duration.ofMillis(5)).build();// 创建 执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 添加 FileSource 到数据流env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "FileSource").print();// 执行任务env.execute("FileSourceRecordStreamingJob");}
}

4.4 BulkFormat读取文件数据

  • BulkFormat从文件中一次读取一批记录,虽然是最 “底层” 的格式实现,但是提供了优化实现的最大灵活性。
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.FileSourceSplit;
import org.apache.flink.connector.file.src.impl.StreamFormatAdapter;
import org.apache.flink.connector.file.src.reader.BulkFormat;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.time.Duration;/*** 描述:flink集成FileSource & forBulkFileFormat使用 & 流模式* BulkFormat:从文件中一次读取一批记录。 它虽然是最 “底层” 的格式实现,但是提供了优化实现的最大灵活性。** @author 浅夏的猫* @version 1.0.0* @date 2024-02-07 15:30:22*/
public class FileSourceBulkStreamingJob {public static void main(String[] args) throws Exception {//创建 批量读取文件的格式函数,其实底层还是通过对单行文件读取BulkFormat<String, FileSourceSplit> bulkFormat = new StreamFormatAdapter<>(new TextLineInputFormat());// 创建 FileSourceFileSource<String> fileSource = FileSource.forBulkFileFormat(bulkFormat, new Path("D:\\flink\\file_source.txt"))//放开注释则使用流模式,每隔5分钟检查是否有新文件,否则默认使用批模式
//                .monitorContinuously(Duration.ofMillis(5)).build();// 创建 执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 添加 FileSource 到数据流env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "FileSource").print();// 执行任务env.execute("FileSourceBulkStreamingJob");}
}

4.5 使用小结

在上面的示例中,我们使用FileSource方法从指定路径读取文本文件,并将其转换为一个数据流,选择不同的输入格式和解析方式,然后我们调用 print 方法将数据流中的数据打印出来。

05 数据源比较

FileSource 是 Flink 中常用的数据源之一,与其他数据源相比,它具有一些优势和劣势,根据实际情况和需求,可以选择不同的数据源来满足任务的要求。

  • 优势

    • 支持读取大规模的文件数据,适用于大数据处理场景。
    • 支持并行读取和处理,能够充分利用集群资源,提高处理效率。
    • 支持多种文件格式和压缩方式,灵活性强。
  • 劣势

    • 对于小文件的处理效率较低,可能会导致资源浪费和性能下降。
    • 无法实时监控文件的变化,需要手动触发重新读取。

06 总结

FileSource 是 Apache Flink 中用于读取文件数据的重要组件,它能够高效地处理大规模的文件数据,并提供丰富的功能和灵活的用法。通过深入了解 FileSource 的工作原理和用法,可以更好地利用 Flink 来实现大规模数据文件的处理和分析任务。

通过以上详细介绍,可以对 Apache Flink 中的 FileSource 有一个全面的了解,从而更好地应用于实际的数据处理项目中

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

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

相关文章

C# GTS四轴运动控制器实例(固高科技步进电机不带编码器)

注&#xff1a;由于电机不带编码器&#xff0c;无法做home和当前位置信息读取&#xff01; 功能&#xff1a; 三个轴的点位运动&#xff1a;前进后退&#xff0c;并分别显示每个轴的移动脉冲数(可以换算为距离)&#xff01; 开发环境&#xff1a;VS2017 硬件设备&#xff1a;固…

Vue-route核心知识整理

目录 1 相关理解 1.1 对 vue-router 的理解 1.2 对 SPA 应用的理解 1.3 对路由的理解 1.3.1 什么是路由&#xff1f; 1.3.2 路由的分类 2 几个注意点 3 路由的基本使用 4 嵌套 (多级) 路由 5 路由传参 5.1 query 方式传参 5.1.1 跳转路由并携带query参数&#xff0…

【项目】HTTP服务器

HTTP服务器 【项目】HTTP服务器项目介绍背景项目描述技术特点开发环境 网络协议栈HTTP协议特点URI & URL & URNURL格式HTTP请求与响应请求响应 CGI机制CGI的实现CGI的意义 日志封装TcpServer类线程池任务类CallBack回调方法类线程池类 封装HttpServer类主函数 封装HTTP…

Python asyncio的理解与入门

asyncio是Python标准库中用于编写异步代码的模块&#xff0c;它提供了一种基于协程的方式来处理并发任务。使用asyncio可以编写高效的异步程序&#xff0c;利用事件循环在单个线程中管理多个任务的执行。 下面是asyncio的基本概念和入门示例&#xff1a; 基本概念&#xff1a…

C 编译生成静态库、动态库的方式

背景 此前编程涉及到自定义编译生成静态库或动态库&#xff0c;为了方便以后查阅&#xff0c;在此记录一下&#xff0c;同时分享给大家。 将C文件编译成静态库或动态库可以使用不同的工具和编译选项&#xff0c;取决于你的需求和平台&#xff1b;以下是使用GCC编译器的简单示例…

Linux 内存top命令详解

通过top命令可以监控当前机器的内存实时使用情况&#xff0c;该命令的参数解释如下&#xff1a; 第一行 15:30:14 —— 当前系统时间 up 1167 days, 5:02 —— 系统已经运行的时长&#xff0c;格式为时:分 1 users ——当前有1个用户登录系统 load average: 0.00, 0.01, 0.05…

【ArcGIS微课1000例】0105:三维模型转体模型(导入sketchup转多面体为例)

文章目录 一、实验概述二、三维模型转多面体三、加载多面体数据四、注意事项一、实验概述 ArcGIS可以借助【导入3D文件】工具支持主流的三维模型导入。支持 3D Studio Max (.3ds)、VRML and GeoVRML 2.0 (.wrl)、SketchUp 6.0 (.skp)、OpenFlight 15.8 (.flt)、Collaborative …

【Web】没人比我更懂SSRF之入门必须要懂的知识汇总

目录 SSRF速览 伪协议 总览 file伪协议 dict伪协议 http协议 gopher伪协议 绕过 127限制绕过 302重定向绕过 DNS重绑定绕过 利用 gopher模拟请求打内网其他漏洞 Gopherus乱杀打组件 先看这篇文章&#xff1a;https://z3r4y.blog.csdn.net/article/details/135458…

算法——排序算法

目录 1、冒泡排序 2、插入排序 3、选择排序 4、归并排序 5、快速排序 6、堆排序 7、计数排序 8、桶排序 9、基数排序 常见的排序算法包括&#xff1a; 冒泡排序&#xff08;Bubble Sort&#xff09;插入排序&#xff08;Insertion Sort&#xff09;选择排序&#xff08;Se…

Oracle12cR2之Job定时作业调度器详解

Oracle12cR2之Job定时作业调度器详解 文章目录 Oracle12cR2之Job定时作业调度器详解1.Oracle Job1. 关于Job2. 使用方法 2. Job详细说明1. 查看Job的相关视图2.SYS.DBA_JOBS视图字段详细说明 3. 创建及查看Job1. 创建Job2. 查看运行中的Job 1.Oracle Job 1. 关于Job 在 Oracle…

关于open3d的 kdtree模块的 查询偶发性崩溃问题

主要问题还是open3d的 kdtree模块的 query相关的算法引起的。没来得及在github提issue也查不到相关issue&#xff0c;因为是偶发性崩溃。所以需要记录一下&#xff0c;防止后人踩坑。 初始化kdtree的方式如下。 sample_pcd_data o3d.data.PCDPointCloud() pcd o3d.io.read_p…

面向对象分析和设计

面向对象分析与设计 面向对象的基本概念 软件工程学家Coad和Yourdon给出了一个定义&#xff1a;面向对象&#xff08;Object-Oriented&#xff09;对象&#xff08;Object&#xff09;继承&#xff08;Inheritance&#xff09;通信&#xff08;Communication&#xff09;。如…

设计模式: 策略模式

文章目录 一、什么是策略模式二、策略模式结构三、使用场景案例分析1、使用场景2、案例分析&#xff08;1&#xff09;消除条件分支 一、什么是策略模式 策略模式是一种行为型设计模式&#xff0c;它允许定义一组算法&#xff0c;并将每个算法封装在独立的类中&#xff0c;使它…

如何设计出用于喜欢的界面

要设计出用户喜欢的界面&#xff0c;你可以考虑以下几个方面&#xff1a; 用户研究&#xff1a;首先要了解用户的需求和偏好。你可以通过用户调研、用户访谈和数据分析来获取这些信息。了解用户的行为模式、喜好和痛点&#xff0c;有助于设计出更吸引人的界面。 直观的布局&am…

LeetCode 2656.K个元素的最大和

给你一个下标从 0 开始的整数数组 nums 和一个整数 k 。你需要执行以下操作 恰好 k 次&#xff0c;最大化你的得分&#xff1a; 从 nums 中选择一个元素 m 。 将选中的元素 m 从数组中删除。 将新元素 m 1 添加到数组中。 你的得分增加 m 。 请你返回执行以上操作恰好 k 次后…

easyx 枪声模拟器

作品介绍:枪声模拟器 简介: “枪声模拟器”是一个基于Windows平台的简单程序,它使用C++编写,主要目的是通过模拟枪声来增强用户的体验。程序使用了图形库来展示一个蓝色的背景屏幕,并提示用户等待片刻后按空格键模拟开枪。当用户按下空格键时,程序会播放预先设定的枪声音…

代码随想录算法训练营第十八天|235.二叉搜索树的最近公共祖先,701.二叉搜索树中的插入操作,450.删除二叉搜索树节点

235.二叉搜索树的最近公共祖先 701.二叉搜索树中的插入操作 450.删除二叉搜索树节点 235.二叉搜索树的最近公共祖先 给定一个二叉搜索树, 找到该树中两个指定节点的最近公共祖先。 百度百科中最近公共祖先的定义为&#xff1a;“对于有根树 T 的两个结点 p、q&#xff0c;最近…

[Swift]定义一个全局的可管理的计时器

定义一个全局计时器&#xff0c;延迟执行操作。可以对计时器进行全局控制&#xff0c;能手动控制暂停/重启/停止&#xff0c;并在计时结束后释放掉计时器。 import Foundationclass TimerManager {static let shared TimerManager()private var timer: DispatchSourceTimer?…

基于Java+Springboot+vue体育用品销售商城平台设计和实现

基于JavaSpringbootvue体育用品销售商城平台设计和实现 博主介绍&#xff1a;5年java开发经验&#xff0c;专注Java开发、定制、远程、文档编写> 博主介绍&#xff1a;5年java开发经验&#xff0c;专注Java开发、定制、远程、文档编写指导等,csdn特邀作者、专注于Java技术领…

在 Vue 中将 DOM 导出为图片

你好&#xff0c;我是小白Coding日志&#xff0c;一个热爱技术的程序员。在这里&#xff0c;我分享自己在编程和技术世界中的学习心得和体会。希望我的文章能够给你带来一些灵感和帮助。欢迎来到我的博客&#xff0c;一起在技术的世界里探索前行吧&#xff01; 在日常的工作中&…