从源码看Spark读取Hive表数据小文件和分块的问题
摘要:使用Spark进行数据分析和计算早已成趋势,你是否关注过读取一张Hive表时Task数为什么是那么多呢?它跟什么有关系呢? 最近刚好碰到这个问题,而之前对此有些模糊,所以做了些整理,希望大家拍砖探讨
前言
有同事问到,Spark读取一张Hive表的数据Task有一万多个,看了Hive表分区下都是3MB~4MB的小文件,每个Task只处理这么小的文件,实在浪费资源浪费时间。而我们都知道Spark的Task数由partitions决定,所以他想通过repartition(num)的方式来改变分区数,结果发现读取文件的时候Task数并没有改变。遂问我有什么参数可以设置,从而改变读取Hive表时的Task数,将小文件合并大文件读上来
> 本文涉及源码基于Spark2.0.0和Hadoop2.6.0,不同版本代码可能不一致,需自己对应。此外针对TextInputFormat格式的Hive表,其他格式的比如Parquet有Spark自己的高效实现,不在讨论范围之内
分析
Spark读取Hive表是通过HadoopRDD扫描上来的,具体可见 org.apache.spark.sql.hive.TableReader类,构建HadoopRDD的代码如下
这里inputFormatClass是Hive创建时指定的,默认不指定为 org.apache.hadoop.mapred.TextInputFormat,由它就涉及到了HDFS文件的FileSplit数,从而决定了上层Spark的partition数。在进入HadoopRDD类查看之前,还有一个参数需要我们注意,就是 _minSplitsPerRDD,它在后面SplitSize的计算中是起了作用的。
我们看一下它的定义
在我们指定以--master local模式跑的时候,它为0,而在其他模式下,则是求的一个最大值。这里重点看 defaultMinPartitions,如下
从这里可以看到,defaultMinPartitions的值一般为2,而 mapred.map.tasks 或者 mapreduce.job.maps( 新版参数)是Hadoop的内建参数,其默认值也为2,一般很少去改变它。所以这里_minSplitsPerRDD的值基本就是2了。
下面我们跟到HadoopRDD类里,去看看它的partitions是如何来的
在 getPartitions 方法里我们可以看到 FileSplit数最后决定了Spark读取Hive表的Task数,下面我们再来看看mapred.TextInputFormat 类里 getSplits 的实现
分两步来看,首先是扫描文件,计算文件大小的部分
针对Hive表的分区,Spark对每个分区都构建了一个HadoopRDD,每个分区目录下就是实际的数据文件,例如我们集群的某一张表按天分区,每天下面有200个数据文件,每个文件大概3MB~4MB之间,这些实际上是reduce设置不合理导致的小文件产生,如下图
此处 listStatus 方法就是扫描的分区目录,它返回的就是图中显示的具体 part-*****文件的FileStatus对象,一共200个。从 totalSize 的计算可以看出,它是这200个文件的总大小,为838MB,因此 goalSize 就为419MB。
参数 mapreduce.input.fileinputformat.split.minsize 在Spark程序没有配的情况下,获取的值为0,而 minSplitSize在Spark获取FileSplits的时候并没有被设置,所以为默认值1,那么 minSize 就为1
其次,我们再来看从文件划分Split,部分代码如下(部分解释见注释)
从上面可以看到,splitSize是从 computeSplitSize(goalSize, minSize, blockSize) 计算来的,这三个参数我们都知道大小,那么计算规则是怎么样的呢。
规则:Math.max(minSize, Math.min(goalSize, blockSize)),从而我们可以知道 splitSize = 128MB,对于3MB~4MB的小文件来说,就 决定了一个小文件就是一个split了,从而对应了一个Spark的partition,所以我们一个分区下就有200个partition,当取两个月的数据时,就是 200 * 30 * 2 = 12000,从而是12000个Task,跟同事所说的吻合!
> 而从TextInputFormat里分Split的逻辑来看,它只会把一个文件分得越来越小,而不会对小文件采取合并,所以无论调整哪个参数,都没法改变这种情况!而通过repartition强行分区,也是在拿到HDFS文件之后对这12000个partition进行重分区,改变不了小文件的问题,也无法改变读取Hive表Task数多的情况。
总结
Block是物理概念,而Split是逻辑概念,最后数据的分片是根据Split来的。一个文件可能大于BlockSize也可能小于BlockSize,大于它就会被分成多个Block存储到不同的机器上,SplitSize可能大于BlockSize也可能小于BlockSize,SplitSize如果大于BlockSize,那么一个Split就可能要跨多个Block。对于数据分隔符而言,不用担心一个完整的句子分在两个Block里,因为在Split构建RecordReader时,它会被补充完整。
对于采用 org.apache.hadoop.mapred.TextInputFormat 作为InputFormat的Hive表,如果存在小文件,Spark在读取的时候单凭调参数和repartition是改变不了分区数的!对于小文件的合并,目前除了Hadoop提供的Archive方式之外,也只能通过写MR来手动合了,最好的方式还是写数据的时候自己控制reduce的个数,把握文件数。
对于Spark直接通过SparkContext的 textFile(inputPath, numPartitions) 方法读取HDFS文件的,它底层也是通过HadoopRDD构建的,它的参数numPartitions就是上面计算goalSize的numSplits参数,这篇 文章 对原理描述的非常详细,非常值得一读。
对于小文件合并的InputFormat有 org.apache.hadoop.mapred.lib.CombineFileInputFormat,跟它相关的参数是 mapreduce.input.fileinputformat.split.maxsize,它用于设置一个Split的最大值。
跟 mapred.TextInputFormat 里的Split划分相关的参数。
mapreduce.input.fileinputformat.split.minsize : 决定了计算Split划分时的minSize。
mapreduce.job.maps 或 mapred.map.tasks : 决定了getSplits(JobConf job, int numSplits)方法里的numSplits,从而可以影响goalSize的大小。
dfs.block.size 或 fs.local.block.size : 决定了HDFS的BlockSize。
MapReduce新版API里的 org.apache.hadoop.mapreduce.lib.input.TextInputFormat,它的SplitSize与上面说到的计算方式不一样,getSplits方法的签名为 getSplits(JobContext job),不再有numSplilts这个参数,splitSize的计算规则改为 Math.max(minSize, Math.min(maxSize, blockSize)),minSize和blockSize跟之前一样,新的maxSize为conf.getLong("mapreduce.input.fileinputformat.split.maxsize", Long.MAX_VALUE)。
在Spark2.0.0里,设置Hadoop相关的参数(比如mapreduce开头的)要通过 spark.sparkContext.hadoopConfiguration 来设置。
感谢观看!
end
网络大数据
(ID:raincent_com)
网络大数据 www.raincent.com
由清华CDN实验室于2011年创办,
致力于打造中国最专业的网络大数据科学门户网站。
识别二维码,关注网络大数据
- 原高价版益盟操盘手经典指标用法及源码分享
- 如果收到外星人发来的信息,人类是否应该不读取反而删除信息?
- iOS 9 的部分源码,是这样被泄露的!
- iOS 9 的源码是这样被泄露的……
- iPhone源码究竟是如何泄露的?一起来了解下
- 程序员的情人节应该这么优雅度过(附源码)
- iOS 9 部分源码遭泄露,苹果已证实
- 苹果史上最严重的泄密!iOS 9 的 iPhone 源码泄露
- JUnit 源码解析
- 程序员表白程序,开放源码在此!