解决Spark流处理产生的小文件问题

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

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

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

图片

1、目的

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

2、影响

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或者对象储存中,小文件的读写处理速度要远远小于大文件,(寻址耗时)。

3、解决思路

3.1、事前

(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的情况。

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

(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);

该特性有两个优点:

(1)通过减少被写入的表文件数量,提高写数据的吞吐量;

(2)避免小文件的产生,提升查询性能。

缺点:

其缺点也是显而易见的,由于使用了一个 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'

4、附加

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

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

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

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

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

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

相关推荐

  1. Spark产生文件原因及解决方案

    2024-06-10 15:08:01       9 阅读
  2. 前端接收后端传文件并下载解决乱码问题

    2024-06-10 15:08:01       34 阅读
  3. Python 处理样本数据文档分类问题

    2024-06-10 15:08:01       22 阅读
  4. Hive文件处理

    2024-06-10 15:08:01       5 阅读
  5. 使用Apache Spark处理Excel文件简易指南

    2024-06-10 15:08:01       28 阅读

最近更新

  1. Aeron:Online Resources

    2024-06-10 15:08:01       0 阅读
  2. 安卓gradel下载失败解决方案

    2024-06-10 15:08:01       0 阅读
  3. 实战

    实战

    2024-06-10 15:08:01      0 阅读
  4. 力扣719.找出第K小的数对距离

    2024-06-10 15:08:01       0 阅读
  5. 10月,2024北京养老展,北京老年生活用品展览会

    2024-06-10 15:08:01       0 阅读
  6. python之面向对象编程

    2024-06-10 15:08:01       0 阅读

热门阅读

  1. python使用关联规则来分析数据

    2024-06-10 15:08:01       7 阅读
  2. 【蓝桥杯】C语言常见高级算法

    2024-06-10 15:08:01       7 阅读
  3. 【leetcode】hot100 哈希表

    2024-06-10 15:08:01       5 阅读
  4. C-Linux: 题集

    2024-06-10 15:08:01       6 阅读
  5. 【DevOps】开源的sd-wan软件介绍和对比分析

    2024-06-10 15:08:01       7 阅读
  6. 词向量对模型performance的影响

    2024-06-10 15:08:01       6 阅读
  7. Linux磁盘与文件系统管理

    2024-06-10 15:08:01       7 阅读
  8. 需求记录(共享元素)

    2024-06-10 15:08:01       9 阅读
  9. Vue3 组合式 API:依赖注入(四)

    2024-06-10 15:08:01       3 阅读
  10. 贪心算法03(leetcode1005,134,135)

    2024-06-10 15:08:01       6 阅读
  11. web前端三大主流框架

    2024-06-10 15:08:01       4 阅读
  12. C++中的if constexpr

    2024-06-10 15:08:01       6 阅读
  13. 驱动开发MISC 杂项驱动

    2024-06-10 15:08:01       7 阅读
  14. 自建 Docker 镜像

    2024-06-10 15:08:01       5 阅读