SparkStreaming如何解决小文件问题

在使用sparkstreaming进行实时计算并将结果写入hdfs时，常见的问题是会产生大量的小文件。这是由于sparkstreaming的微批处理模式和dstream（rdd）的分布式（partition）特性所导致的。每个partition会启动一个独立的线程来处理数据，导致每个batch的每个partition都会生成一个新的文件流。假设一个batch为10秒，每个输出的dstream有32个partition，那么一小时内产生的文件数量将达到(3600/10)*32=11520个。如此众多的小文件会给namenode带来巨大的压力，因为需要维护大量的文件元信息，如文件位置、文件大小、block数量等。无论是parquet、text、json还是avro格式的文件，都会遇到这种小文件问题。以下是几种处理sparkstreaming小文件的典型方法：

增加batch大小

这种方法非常简单易懂，batch越大，从外部接收的事件就越多，内存中积累的数据也越多，输出的文件数量自然会减少。例如，将batch时间从10秒增加到100秒，那么一小时的文件数量就会减少到1152个。然而，这种方法的缺点是实时业务可能无法接受如此长的延迟。本来10秒就能看到结果更新一次，现在要等将近两分钟，用户可能会不满。因此，这种方法适用于消息实时到达但不想挤压在一起处理的场景，因为挤压在一起处理会导致批处理任务等待，这时可以采用这种方法（这有点像Spark内部的pipeline模式，但要注意两者的区别）。

Coalesce大法好？

文章开头提到，小文件的数量由batch_number和partition_number决定，第一种方法是减少batch_number，而这种方法则是减少partition_number。这个API无需详细说明，就是减少初始的分区数量。熟悉Spark源码的朋友都知道，对于窄依赖，一个子RDD的partition规则继承自父RDD，而对于宽依赖（如那些以ByKey结尾的操作），如果没有特别指定分区数量，也会继承自父RDD。因此，初始的SourceDstream有几个partition，最终的输出就有几个partition。Coalesce大法的优势在于，可以在最终输出时减少partition数量。然而，这种方法的缺点也很明显，原本是32个线程在写256M数据，现在可能变成了4个线程在写256M数据，而如果没有完成这256M数据的写入，这个batch就不会结束。因此，一个batch的处理延迟必然会增加，batch挤压问题会逐渐加剧。这种方法也需要谨慎使用，切记要根据实际情况来决定。

SparkStreaming外部处理

既然我们将数据输出到HDFS，说明肯定是要使用Hive或SparkSQL这样的“SQL on Hadoop”系统进行进一步的数据分析，而这些表通常是按半小时、一小时或一天来分区的（注意不要与SparkStreaming的分区混淆，这里说的分区是用于分区裁剪优化的）。因此，我们可以考虑在SparkStreaming之外启动定时的批处理任务来合并SparkStreaming产生的小文件。这种方法虽然不是很直接，但却非常有用，性价比高。唯一需要注意的是，批处理的合并任务在时间切割上要把握好，否则可能会合并到还在写入的SparkStreaming小文件。

文小言

百度旗下新搜索智能助手，有问题，问小言。

57

查看详情

自己调用foreach去追加

SparkStreaming提供了foreach这个output类API，可以让我们自定义输出计算结果的方法。我们可以利用这个特性，每个batch在写文件时，不是生成一个新的文件流，而是打开之前的文件。考虑这种方法的可行性，首先，HDFS上的文件不支持修改，但许多文件支持追加。因此，每个batch的每个partition可以对应一个输出文件，每次都追加到这个partition对应的输出文件，这样也可以减少文件数量。这种方法需要注意的是不能无限制地追加，当判断一个文件已经达到某个阈值时，就要生成一个新的文件进行追加。

我已经尝试过上述所有方法，各有优劣，大家在使用时需多加注意。

以上就是SparkStreaming如何解决小文件问题的详细内容，更多请关注php中文网其它相关文章！