Hive中如何确定map数

2013.09.04 | Comments

Hive 是基于 Hadoop的一个数据仓库工具,可以将结构化的数据文件映射为一张数据库表,并提供完整的 sql 查询功能,可以将 sq l语句转换为 MapReduce 任务进行运行。当运行一个 hql 语句的时候,map 数是如何计算出来的呢?有哪些方法可以调整 map 数呢?

本文测试集群版本:cdh-4.3.0

hive 默认的 input format

cdh-4.3.0 的 hive 中查看 hive.input.format 值(为什么是hive.input.format?):

hive> set hive.input.format;
hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;

可以看到默认值为 CombineHiveInputFormat,如果你使用的是 IDH 的hive,则默认值为:

hive> set hive.input.format;
hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;

CombineHiveInputFormat 类继承自 HiveInputFormat,而 HiveInputFormat 实现了 org.apache.hadoop.mapred.InputFormat 接口,关于 InputFormat 的分析,可以参考Hadoop深入学习:InputFormat组件.

InputFormat 接口功能

简单来说,InputFormat 主要用于描述输入数据的格式,提供了以下两个功能:

1)、数据切分,按照某个策略将输入数据且分成若干个 split,以便确定 Map Task 的个数即 Mapper 的个数,在 MapReduce 框架中,一个 split 就意味着需要一个 Map Task;

2)、为 Mapper 提供输入数据,即给定一个 split(使用其中的 RecordReader 对象)将之解析为一个个的 key/value 键值对。

该类接口定义如下:

public interface InputFormat<K,V>{
	public InputSplit[] getSplits(JobConf job,int numSplits) throws IOException; 
	public RecordReader<K,V> getRecordReader(InputSplit split,JobConf job,Reporter reporter) throws IOException; 
}

其中,getSplit() 方法主要用于切分数据,每一份数据由,split 只是在逻辑上对数据分片,并不会在磁盘上将数据切分成 split 物理分片,实际上数据在 HDFS 上还是以 block 为基本单位来存储数据的。InputSplit 只记录了 Mapper 要处理的数据的元数据信息,如起始位置、长度和所在的节点。

MapReduce 自带了一些 InputFormat 的实现类:

InputFormat实现类

hive 中有一些 InputFormat 的实现类,如:

AvroContainerInputFormat
RCFileBlockMergeInputFormat
RCFileInputFormat
FlatFileInputFormat
OneNullRowInputFormat
ReworkMapredInputFormat
SymbolicInputFormat
SymlinkTextInputFormat
HiveInputFormat

HiveInputFormat 的子类有:

HiveInputFormat的子类

HiveInputFormat

以 HiveInputFormat 为例,看看其getSplit()方法逻辑:

for (Path dir : dirs) {
  PartitionDesc part = getPartitionDescFromPath(pathToPartitionInfo, dir);
  // create a new InputFormat instance if this is the first time to see this
  // class
  Class inputFormatClass = part.getInputFileFormatClass();
  InputFormat inputFormat = getInputFormatFromCache(inputFormatClass, job);
  Utilities.copyTableJobPropertiesToConf(part.getTableDesc(), newjob);

  // Make filter pushdown information available to getSplits.
  ArrayList<String> aliases =
      mrwork.getPathToAliases().get(dir.toUri().toString());
  if ((aliases != null) && (aliases.size() == 1)) {
    Operator op = mrwork.getAliasToWork().get(aliases.get(0));
    if ((op != null) && (op instanceof TableScanOperator)) {
      TableScanOperator tableScan = (TableScanOperator) op;
      pushFilters(newjob, tableScan);
    }
  }

  FileInputFormat.setInputPaths(newjob, dir);
  newjob.setInputFormat(inputFormat.getClass());
  InputSplit[] iss = inputFormat.getSplits(newjob, numSplits / dirs.length);
  for (InputSplit is : iss) {
    result.add(new HiveInputSplit(is, inputFormatClass.getName()));
  }
}

上面代码主要过程是:

遍历每个输入目录,然后获得 PartitionDesc 对象,从该对象调用 getInputFileFormatClass 方法得到实际的 InputFormat 类,并调用其 getSplits(newjob, numSplits / dirs.length) 方法。

按照上面代码逻辑,似乎 hive 中每一个表都应该有一个 InputFormat 实现类。在 hive 中运行下面代码,可以查看建表语句:

hive> show create table info; 
OK
CREATE  TABLE info(
  statist_date string, 
  statistics_date string, 
  inner_code string, 
  office_no string, 
  window_no string, 
  ticket_no string, 
  id_kind string, 
  id_no string, 
  id_name string, 
  area_center_code string)
ROW FORMAT DELIMITED 
  FIELDS TERMINATED BY '\;' 
  LINES TERMINATED BY '\n' 
STORED AS INPUTFORMAT 
  'org.apache.hadoop.mapred.TextInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION
  'hdfs://node:8020/user/hive/warehouse/info'
TBLPROPERTIES (
  'numPartitions'='0', 
  'numFiles'='1', 
  'transient_lastDdlTime'='1378245263', 
  'numRows'='0', 
  'totalSize'='301240320', 
  'rawDataSize'='0')
Time taken: 0.497 seconds

从上面可以看到 info 表的 INPUTFORMAT 为org.apache.hadoop.mapred.TextInputFormat,TextInputFormat 继承自FileInputFormat。FileInputFormat 是一个抽象类,它最重要的功能是为各种 InputFormat 提供统一的 getSplits()方法,该方法最核心的是文件切分算法和 Host 选择算法。

算法如下:

long length = file.getLen();
long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits);
long minSize = Math.max(job.getLong(org.apache.hadoop.mapreduce.lib.input.
FileInputFormat.SPLIT_MINSIZE, 1), minSplitSize);

long blockSize = file.getBlockSize();
long splitSize = computeSplitSize(goalSize, minSize, blockSize);
long bytesRemaining = length;
while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
String[] splitHosts = getSplitHosts(blkLocations, 
	length-bytesRemaining, splitSize, clusterMap);
	splits.add(makeSplit(path, length-bytesRemaining, splitSize, 
		       splitHosts));
	bytesRemaining -= splitSize;
}

华丽的分割线:以下摘抄自Hadoop深入学习:InputFormat组件

1)文件切分算法

文件切分算法主要用于确定InputSplit的个数以及每个InputSplit对应的数据段,FileInputSplit以文件为单位切分生成InputSplit。有三个属性值来确定InputSplit的个数:

  • goalSize:该值由 totalSize/numSplits 来确定 InputSplit 的长度,它是根据用户的期望的 InputSplit 个数计算出来的;numSplits 为用户设定的 Map Task 的个数,默认为1。
  • minSize:由配置参数 mapred.min.split.size(或者 mapreduce.input.fileinputformat.split.minsize)决定的 InputForma t的最小长度,默认为1。
  • blockSize:HDFS 中的文件存储块block的大小,默认为64MB。
  • numSplits=mapred.map.tasks 或者 mapreduce.job.maps

这三个参数决定一个 InputFormat 分片的最终的长度,计算方法如下:

splitSize = max{minSize,min{goalSize,blockSize}} 

计算出了分片的长度后,也就确定了 InputFormat 的数目。

2)host 选择算法

InputFormat 的切分方案确定后,接下来就是要确定每一个 InputSplit 的元数据信息。InputSplit 元数据通常包括四部分,<file,start,length,hosts>其意义为:

  • file 标识 InputSplit 分片所在的文件;
  • InputSplit 分片在文件中的的起始位置;
  • InputSplit 分片的长度;
  • 分片所在的 host 节点的列表。

InputSplit 的 host 列表的算作策略直接影响到运行作业的本地性。

我们知道,由于大文件存储在 HDFS上的 block 可能会遍布整个 Hadoop 集群,而一个 InputSplit 分片的划分算法可能会导致一个 split 分片对应多个不在同一个节点上的 blocks,这就会使得在 Map Task 执行过程中会涉及到读其他节点上的属于该 Task 的 block 中的数据,从而不能实现数据本地性,而造成更多的网络传输开销。

一个 InputSplit 分片对应的 blocks 可能位于多个数据节点地上,但是基于任务调度的效率,通常情况下,不会把一个分片涉及的所有的节点信息都加到其host列表中,而是选择包含该分片的数据总量的最大的前几个节点,作为任务调度时判断是否具有本地性的主要凭证。

FileInputFormat 使用了一个启发式的 host 选择算法:首先按照 rack 机架包含的数据量对 rack 排序,然后再在 rack 内部按照每个 node 节点包含的数据量对 node 排序,最后选取前 N 个(N 为 block 的副本数),node 的 host 作为 InputSplit 分片的 host 列表。当任务地调度 Task 作业时,只要将 Task 调度给 host 列表上的节点,就可以认为该 Task 满足了本地性。

从上面的信息我们可以知道,当 InputSplit 分片的大小大于 block 的大小时,Map Task 并不能完全满足数据的本地性,总有一本分的数据要通过网络从远程节点上读数据,故为了提高 Map Task 的数据本地性,减少网络传输的开销,应尽量是 InputFormat 的大小和 HDFS 的 block 块大小相同。


CombineHiveInputFormat

getSplits(JobConf job, int numSplits) 代码运行过程如下:

init(job);
CombineFileInputFormatShim combine = ShimLoader.getHadoopShims().getCombineFileInputFormat();
	ShimLoader.loadShims(HADOOP_SHIM_CLASSES, HadoopShims.class);
		Hadoop23Shims
			HadoopShimsSecure.getCombineFileInputFormat()

CombineFileInputFormatShim 继承了org.apache.hadoop.mapred.lib.CombineFileInputFormat,CombineFileInputFormatShim 的 getSplits 方法代码如下:

public InputSplitShim[] getSplits(JobConf job, int numSplits) throws IOException {
  long minSize = job.getLong("mapred.min.split.size", 0);

  // For backward compatibility, let the above parameter be used
  if (job.getLong("mapred.min.split.size.per.node", 0) == 0) {
    super.setMinSplitSizeNode(minSize);
  }

  if (job.getLong("mapred.min.split.size.per.rack", 0) == 0) {
    super.setMinSplitSizeRack(minSize);
  }

  if (job.getLong("mapred.max.split.size", 0) == 0) {
    super.setMaxSplitSize(minSize);
  }

  InputSplit[] splits = (InputSplit[]) super.getSplits(job, numSplits);

  InputSplitShim[] isplits = new InputSplitShim[splits.length];
  for (int pos = 0; pos < splits.length; pos++) {
    isplits[pos] = new InputSplitShim((CombineFileSplit)splits[pos]);
  }

  return isplits;
}

从上面代码可以看出,如果为 CombineHiveInputFormat,则以下四个参数起作用:

  • mapred.min.split.size 或者 mapreduce.input.fileinputformat.split.minsize
  • mapred.max.split.size 或者 mapreduce.input.fileinputformat.split.maxsize
  • mapred.min.split.size.per.rack 或者 mapreduce.input.fileinputformat.split.minsize.per.rack
  • mapred.min.split.size.per.node 或者 mapreduce.input.fileinputformat.split.minsize.per.node

CombineFileInputFormatShim 的 getSplits 方法最终会调用父类的 getSplits 方法,拆分算法如下:

long left = locations[i].getLength();
long myOffset = locations[i].getOffset();
long myLength = 0;
do {
	if (maxSize == 0) {
		myLength = left;
	} else {
	if (left > maxSize && left < 2 * maxSize) {
	  myLength = left / 2;
	} else {
	  myLength = Math.min(maxSize, left);
	}
	}
	OneBlockInfo oneblock = new OneBlockInfo(path, myOffset,
	  myLength, locations[i].getHosts(), locations[i]
	      .getTopologyPaths());
	left -= myLength;
	myOffset += myLength;

	blocksList.add(oneblock);
} while (left > 0);

hive 中如何确定 map 数

总上总结如下:

如果 hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat,则这时候的参数如下:

hive> set mapred.min.split.size;
mapred.min.split.size=1
hive> set mapred.map.tasks;
mapred.map.tasks=2
hive> set dfs.blocksize;
dfs.blocksize=134217728

上面参数中 mapred.map.tasks 为2,dfs.blocksize(使用的是 cdh-4.3.0 版本的 hadoop,这里 block 和 size 之间没有逗号)为128M。

假设有一个文件为200M,则按上面 HiveInputFormat 的 split 算法:

1、文件总大小为200M,goalSize=200M /2 =100M,minSize=1 ,splitSize = max{1,min{100M,128M}} =100M

2、200M / 100M >1.1,故第一块大小为100M

3、剩下文件大小为100M,小于128M,故第二块大小为100M。

如果 hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat,则这时候的参数如下:

hive> set mapred.min.split.size;
mapred.min.split.size=1
hive> set mapred.max.split.size;
mapred.max.split.size=67108864
hive> set mapred.min.split.size.per.rack;
mapred.min.split.size.per.rack=1
hive> set mapred.min.split.size.per.node;
mapred.min.split.size.per.node=1
hive> set dfs.blocksize;
dfs.blocksize=134217728

上面参数中 mapred.max.split.size 为64M,dfs.blocksize 为128M。

假设有一个文件为200M,则按上面 CombineHiveInputFormat 的 split 算法:

1、128M < 200M <128M X 2,故第一个block大小为128M

2、剩下文件大小为200M-128M=72M,72M < 128M,故第二块大小为72M

总结

网上有一些文章关于 hive 中如何控制 map 数的文章是否考虑的不够全面,没有具体情况具体分析。简而言之,当 InputFormat 的实现类为不同类时,拆分块算法都不一样,相关设置参数也不一样,需要具体分析。

1. map 数不是越多越好

如果一个任务有很多小文件(远远小于块大小128m),则每个小文件也会被当做一个块,用一个 map 任务来完成,而一个 map 任务启动和初始化的时间远远大于逻辑处理的时间,就会造成很大的资源浪费。 而且,同时可执行的 map 数是受限的。

2. 如何适当的增加 map 数?

  • 将数据导入到 hive 前,手动将大文件拆分为小文件
  • 指定 map 数,使用 insert 或者 create as select 语句将一个表导入到另一个表,然后对另一张表做查询

3. 一些经验

  • 合并小文件可以减少 map 数,但是会增加网络 IO。

  • 尽量使拆分块大小和 hdfs 的块大小接近,避免一个拆分块大小上的多个 hdfs 块位于不同数据节点,从而降低网络 IO。

  • 根据实际情况,控制 map 数量需要遵循两个原则:使大数据量利用合适的map数使单个map任务处理合适的数据量。

参考文章


原创文章,转载请注明: 转载自JavaChen Blog,作者:JavaChen
本文链接地址:http://blog.javachen.com/2013/09/04/how-to-decide-map-number.html
本文基于署名2.5中国大陆许可协议发布,欢迎转载、演绎或用于商业目的,但是必须保留本文署名和文章链接。 如您有任何疑问或者授权方面的协商,请邮件联系我。