Spark 小文件合并

1. 问题描述

最近使用 spark sql 执行 etl 时候出现了,最终结果大小只有几百 k,但是小文件一个分区有上千的情况。

危害:

  1. hdfs 有最大文件数限制
  2. 浪费磁盘资源(可能存在空文件);
  3. hive 中进行统计, 计算的时候, 会产生很多个 map, 影响计算的速度。

2. 解决方法

  1. 方法一:通过 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 的情况。

    但是由于使用的是同事直接写好的模块,改新增函数相对比较麻烦,所以作为后手。

  2. 方法二:降低 spark 并行度,即调节 spark.sql.shuffle.partitions

    比如之前设置的为 100,按理说应该生成的文件数为 100;
    但是由于业务比较特殊,采用的大量的 union all,且 union all 在 spark 中属于窄依赖,
    不会进行 shuffle,所以导致最终会生成(union all 数量 +1)*100 的文件数。
    如有 10 个 union all,会生成 1100 个小文件。
    这样导致降低并行度为 10 之后,执行时长大大增加,且文件数依旧有 110 个,效果有,但是不理想。

  3. 方法三:新增一个并行度 =1 任务,专门合并小文件。

    先将原来的任务数据写到一个临时分区(如 tmp);
    再起一个并行度为 1 的任务,类似:

    insert overwrite 目标表 select * from 临时分区

    但是结果小文件数还是没有减少,略感疑惑;
    经过多次测后发现原因:‘select * from 临时分区’ 这个任务在 spark 中属于窄依赖;
    并且 spark DAG 中分为宽依赖和窄依赖,只有宽依赖会进行 shuffle;
    故并行度 shuffle,spark.sql.shuffle.partitions=1 也就没有起到作用;

    由于数据量本身不是特别大,所以直接采用了 group by(在 spark 中属于宽依赖)的方式,类似:

    insert overwrite 目标表 select * from 临时分区 group by *

    先运行原任务,写到 tmp 分区,‘dfs -count’查看文件数,1100 个,运行加上 group by 的临时任务(spark.sql.shuffle.partitions=1),查看结果目录,文件数 =1,成功。
    最后又加了个删除 tmp 分区的任务。

3. 结论

  1. 方便的话,可以采用 coalesce()方法和 repartition() 方法

  2. 如果任务逻辑简单,数据量少,可以直接降低并行度

  3. 任务逻辑复杂,数据量很大,原任务大并行度计算写到临时分区,再加两个任务:
    一个用来将临时分区的文件用小并行度(加宽依赖)合并成少量文件到实际分区;
    另一个删除临时分区;

  4. hive 任务减少小文件相对比较简单,可以直接设置参数,如:

    Map-only 的任务结束时合并小文件:
    sethive.merge.mapfiles = true

    在 Map-Reduce 的任务结束时合并小文件:
    sethive.merge.mapredfiles= true

    当输出文件的平均大小小于 1GB 时,启动一个独立的 map-reduce 任务进行文件 merge:
    sethive.merge.smallfiles.avgsize=1024000000

另外参见:https://itbook.com/article/1598106371620