合并spark structured streaming处理流式数据产生的小文件

news2025/2/28 13:03:56

备注:

By 远方时光原创,可转载,不能复制到其他平台

背景:做流批一体,湖仓一体的大数据架构,常见的做法就是

数据源->spark Streaming->ODS(数据湖)->spark streaming->DWD(数据湖)->...

那么数据源->spark Streaming->ODS,以这段为例,在数据源通过spark structured streaming写入ODS在数据湖(Delta Lake)落盘时候必然会产生很多小文件

目的:

为了在批处理spark-sql运行更快,也避免因为小文件而导致报错

影响:

WARNING: Failed to connect to /172.16.xx.xx:9866 for block, add to deadNodes and continue. java.net.SocketException: Too many open files
 

1.小文件在批处理数据IO消耗巨大,程序可能卡死

2.小文件块都有对应的元数据,元数据放在NameNode,导致需要的内存大大增大,增加NameNode压力,这样会限制了集群的扩展。

3.在HDFS或者对象储存中,小文件的读写处理速度要远远小于大文件,(寻址耗时)

解决思路:

事前:

1.避免写入时候产生过多小文件

  • 做好分区partitionBy(年,月,日), 避免小文件过于分散
  • Trigger触发时间可以设置为1分钟,这样会攒一批一写入,避免秒级别写入而产生大量小文件(但是使用spark structured 想要做real-time不能这样,只适合做准实时

2.打开自适应框架的开关

spark.sql.adaptive.enabled true

3.通过spark的coalesce()方法和repartition()方法

val rdd2 = rdd1.coalesce(8, true) //(true表示是否shuffle)
val rdd3 = rdd1.repartition(8)

coalesce:coalesce()方法的作用是返回指定一个新的指定分区的Rdd,如果是生成一个窄依赖的结果,那么可以不发生shuffle,分区的数量发生激烈的变化,计算节点不足,不设置true可能会出错。
repartition:coalesce()方法shuffle为true的情况。

事后(小文件引起已经产生):

1:优化 Delta 表的写入,避免小文件产生

在开源版 Spark 中,每个 executor 向 partition 中写入数据时,都会创建一个表文件进行写入,最终会导致一个 partition 中产生很多的小文件。

Databricks 对 Delta 表的写入过程进行了优化,对每个 partition,使用一个专门的 executor 合并其他 executor 对该 partition 的写入,从而避免了小文件的产生。

该特性由表属性 delta.autoOptimize.optimizeWrite 来控制:

可以在创建表时指定

CREATE TABLE student (id INT, name STRING)
TBLPROPERTIES (delta.autoOptimize.optimizeWrite = true);

也可以修改表属性

ALTER TABLE table_name
SET TBLPROPERTIES (delta.autoOptimize.optimizeWrite = true);

该特性有两个优点:通过减少被写入的表文件数量,提高写数据的吞吐量;避免小文件的产生,提升查询性能。

其缺点也是显而易见的,由于使用了一个 executor 来合并表文件的写入,从而降低了表文件写入的并行度,此外,多引入的一层 executor 需要对写入的数据进行 shuffle,带来额外的开销。因此,在使用该特性时,需要对场景进行评估:

该特性适用的场景:频繁使用 MERGE,UPDATE,DELETE,INSERT INTO,CREATE TABLE AS SELECT 等 SQL 语句的场景;

该特性不适用的场景:写入 TB 级以上数据。

2.自动合并小文件

在流处理场景中,比如流式数据入湖场景下,需要持续的将到达的数据插入到 Delta 表中,每次插入都会创建一个新的表文件用于存储新到达的数据,假设每10s触发一次,那么这样的流处理作业一天产生的表文件数量将达到8640个,且由于流处理作业通常是 long-running 的,运行该流处理作业100天将产生上百万个表文件。这样的 Delta 表,仅元数据的维护就是一个很大的挑战,查询性能更是急剧恶化。

为了解决上述问题,Databricks 提供了小文件自动合并功能,在每次向 Delta 表中写入数据之后,会检查 Delta 表中的表文件数量,如果 Delta 表中的小文件(size < 128MB 的视为小文件)数量达到阈值,则会执行一次小文件合并,将 Delta 表中的小文件合并为一个新的大文件。

该特性由表属性 delta.autoOptimize.autoCompact 控制,和特性 delta.autoOptimize.optimizeWrite 相同,可以在创建表时指定,也可以对已创建的表进行修改。自动合并的阈值由 spark.databricks.delta.autoCompact.minNumFiles 控制,默认为50,即小文件数量达到50会执行表文件合并;合并后产生的文件最大为128MB,如果需要调整合并后的目标文件大小,可以通过调整配置 spark.databricks.delta.autoCompact.maxFileSize 实现。

3:手动合并小文件(我常用,每天定时运行合并分区内小文件,再去处理批任务)

自动小文件合并会在对 Delta 表进行写入,且写入后表中小文件达到阈值时被触发。除了自动合并之外,Databricks 还提供了 Optimize 命令使用户可以手动合并小文件,优化表结构,使得表文件的结构更加紧凑。在实现上 Optimize 使用 bin-packing 算法,该算法不但会合并表中的小文件,且合并后生成的表文件也更均衡(表文件大小相近)。例如,我们要对 Delta 表 student 的表文件进行优化,仅需执行如下命令即可实现:(Optimize 命令不但支持全表小文件的合并,还支持特定的分区的表文件的合并

OPTIMIZE student WHERE date >= '2024-01-01'

附加:

面试官可能会问,我运行optimize合并小文件,但是小文件太多了,直接卡死运行不了程序(某互联网面试题)

回答:

1.首先停掉程序,这里注意deltalake因为有历史版本这个概念,所以不存在运行一半覆盖原来版本情况,可以基于上一个版本重新运行(考点)

2.第二点,大数据思想分而治之“分”,即把复杂的任务分解为若干个“简单的任务”来处理。

​OPTIMIZE student WHERE date > '2024-01-01' and date < '2024-01-02'

因为前面做了partitionby(年月日),那么缩小optimize范围,在遍历这个月的每一天日期,分治处理

3.第三点,大数据思想,自己不行找兄弟,加节点,加计算资源

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

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

相关文章

如何让网页APP化 渐进式Web应用(PWA)

前言 大家上网应该发现有的网页说可以安装对应应用&#xff0c;结果这个应用好像就是个web&#xff0c;不像是应用&#xff0c;因为这里采用了PWA相关技术。 PWA&#xff0c;全称为渐进式Web应用&#xff08;Progressive Web Apps&#xff09;&#xff0c;是一种可以提供类似…

无法访问云服务器上部署的Docker容器(二)

说明&#xff1a;记录一次使用公网IP 接口地址无法访问阿里云服务接口的问题&#xff1b; 描述 最近&#xff0c;我使用Docker部署了jeecg-boot项目&#xff0c;部署过程都没有问题&#xff0c;也没有错误信息。部署完成后&#xff0c;通过下面的地址访问后端Swagger接口文档…

Facebook的虚拟社交愿景:元宇宙时代的新起点

在当今数字化时代&#xff0c;社交媒体已经成为人们生活中不可或缺的一部分。而随着科技的不断进步和社会的发展&#xff0c;元宇宙已经成为了人们关注的热点话题之一。作为社交媒体的领军企业之一&#xff0c;Facebook也在积极探索虚拟社交的未来&#xff0c;将其视为元宇宙时…

微服务-微服务链路追踪组件Skywalking实战

自动化监控系统Prometheus&Grafana实战&#xff1a; 4 trem APM-性能监控项目班&#xff1a; https://vip.tulingxueyuan.cn/detail/p_602e574ae4b035d3cdb8f8fe/6 1. skywalking是什么 1.1 Skywalking主要功能特性 1.2 Skywalking整体架构 1.3 SkyWalking 环境搭建部…

【Linux C | 网络编程】套接字选项、getsockopt、setsockopt详解及C语言例子

&#x1f601;博客主页&#x1f601;&#xff1a;&#x1f680;https://blog.csdn.net/wkd_007&#x1f680; &#x1f911;博客内容&#x1f911;&#xff1a;&#x1f36d;嵌入式开发、Linux、C语言、C、数据结构、音视频&#x1f36d; &#x1f923;本文内容&#x1f923;&a…

Prompt 编程的优化技巧

一、为什么要优化 一&#xff09;上下文限制 目前 GPT-3.5 以及 GPT-4最大支持 16K 上下文&#xff0c;比如你输入超过 16k 的长文本&#xff0c;ChatGPT 会提示文本过大&#xff0c;为了避免 GPT 无法回复&#xff0c;需要限制 上下文在16k 以内 上下文对于 GPT 来说是非常重…

【Java代码审计】SSRF

什么是SSRF&#xff1f; SSRF(Server-Side Request Forge, 服务端请求伪造) &#xff0c;即攻击者构造恶意参数使服务端对其它内/外网系统进行访问或者攻击的一种方式。 Java支持的网络请求协议&#xff1a; file ftp http https gopher(jdk≤1.7) jar netdoc mailto SSRF代码…

接口的定义与实现方法

作用&#xff1a; 约束 定义一些方法&#xff0c;让不同的人实现 public abstract piblic static final 接口不能被实例化&#xff0c;接口中没有构造方法 implements可以实现多个接口 必须要重写接口中的方法~

Vue前端对请假模块——请假开始时间和请假结束时间的校验处理

开发背景&#xff1a;Vueelement组件开发 业务需求&#xff1a;用户提交请假申请单&#xff0c;请假申请的业务逻辑处理 实现&#xff1a;用户选择开始时间需要大于本地时间&#xff0c;不得大于请假结束时间&#xff0c;请假时长根据每日工作时间实现累加计算 页面布局 在前…

Mistral发布语言大模型Mistral Large;法国新星Mistral挑战 OpenAI 霸主地位

&#x1f989; AI新闻 &#x1f680; Mistral发布语言大模型Mistral Large 摘要&#xff1a;Mistral Large 是 Mistral AI 公司最新发布的旗舰语言模型&#xff0c;具备顶尖水平的推理能力。它主要被设计用于处理复杂的多语言推理任务&#xff0c;比如文本理解、转换和代码生…

“智能语音指令解析“ 基于NLP与语音识别的工单关键信息提取

“智能语音指令解析“ 基于NLP与语音识别的工单关键信息提取 1. 背景介绍1.1 场景痛点1.2 方案选型 2. 准备开发环境3. PaddleSpeech 语音识别快速使用4. PaddleNLP 信息抽取快速使用5. 语音工单信息抽取核心功能实现6. 语音工单信息抽取网页应用6.1 网页前端6.2 网页后端6.3 a…

3. Java中的锁

文章目录 乐观锁与悲观锁乐观锁(无锁编程,版本号机制)悲观锁两种锁的伪代码比较 通过 8 种锁运行案例,了解锁锁相关的 8 种案例演示场景一场景二场景三场景四场景五场景六场景七场景八 synchronized 有三种应用方式8 种锁的案例实际体现在 3 个地方 从字节码角度分析 synchroni…

ARM 版银河麒麟桌面系统下 Qt 开发环境搭建指南

目录 前言安装Linux ARM 版 QtCreator配置 Qt Creator配置构建套件 第一个麒麟 Qt 应用程序小结 前言 在上一篇文章信创ARM架构QT应用开发环境搭建中建议大家使用 Ubuntu X86 系统作为信创 ARM 架构 QT 应用的开发环境&#xff0c;里面使用了交叉编译的方式。这对于自己的 Qt …

AI与大数据:智慧城市安全的护航者与变革引擎

一、引言 在数字化浪潮的席卷下&#xff0c;智慧城市正成为现代城市发展的新方向。作为城市的神经系统&#xff0c;AI与大数据的融合与应用为城市的安全与应急响应带来了革命性的变革。它们如同城市的“智慧之眼”和“聪明之脑”&#xff0c;不仅为城市管理者提供了强大的决策…

【算法与数据结构】684、685、LeetCode冗余连接I II

文章目录 一、684、冗余连接 I二、685、冗余连接 II三、完整代码 所有的LeetCode题解索引&#xff0c;可以看这篇文章——【算法和数据结构】LeetCode题解。 一、684、冗余连接 I 思路分析&#xff1a;题目给出一个无向有环图&#xff0c;要求去掉一个边以后构成一个树&#xf…

PyQt6的开发流程(密码生成小程序为例)

PyQt6的开发流程&#xff08;密码生成小程序为例&#xff09; 文章目录 PyQt6的开发流程&#xff08;密码生成小程序为例&#xff09;一、流程介绍与概览1. 界面与逻辑分离的开发流程2. PyQt6的开发流程 二、打开 designer.exe 创建文件三、用QT设计师绘制界面保存成ui1. QT常用…

急中生智:献血200cc没事,为啥出血200cc就可能噶?

点击文末“阅读原文”即可参与节目互动 剪辑、音频 / 卷圈 运营 / SandLiu 卷圈 监制 / 姝琦 封面 / 姝琦Midjourney 产品统筹 / bobo 场地支持 / 声湃轩北京录音间 外伤出血更常见&#xff0c;但同样可能危及生命。 众所周知&#xff0c;出血是一种常见的外伤和急症&…

家装服务管理:Java技术的创新应用

✍✍计算机毕业编程指导师 ⭐⭐个人介绍&#xff1a;自己非常喜欢研究技术问题&#xff01;专业做Java、Python、微信小程序、安卓、大数据、爬虫、Golang、大屏等实战项目。 ⛽⛽实战项目&#xff1a;有源码或者技术上的问题欢迎在评论区一起讨论交流&#xff01; ⚡⚡ Java、…

ONLYOFFICE 桌面编辑器 v8.0 更新内容详细攻略

文章目录 引言PDF 表单RTL 支持电子表格中的新增功能Moodle 集成用密码保护 PDF 文件从“开始”菜单快速创建文档本地界面主题下载安装桌面编辑工具总结 引言 官网链接&#xff1a; ONLYOFFICE 官方网址 ONLYOFFICE 桌面编辑器是一款免费的文档处理软件&#xff0c;适用于 Li…

APP被针对攻击了,要怎么解决

随着APP行业的兴起&#xff0c;游戏公司异军突起&#xff0c;不管是在控证还是攻击方面都是属于最复杂的一个场面&#xff0c;游戏APP逐渐成为DDOS流量攻击的“重灾区”。没有提前做好了解就盲目进军游戏APP行业&#xff0c;一旦被攻击就会让公司束手无策。那么&#xff0c;刚上…