hivemap转string_Hive如何确定map数的？

最近批量刷数据的时候&＃xff0c;由于集群资源紧张&＃xff0c;需要控制一些 map 的数量&＃xff0c;本文从底层代码触发&＃xff0c;带大家了解一下 MR 是如何让切分 map 数的。

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

本文测试集群版本&＃xff1a;cdh-4.3.0

在 cdh-4.3.0 的 hive 中查看 hive.input.format 值(为什么是hive.input.format&＃xff1f;)&＃xff1a;

hive> set hive.input.format;hive.input.format&＃61;org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;

可以看到默认值为 CombineHiveInputFormat&＃xff0c;如果你使用的是 IDH 的hive&＃xff0c;则默认值为&＃xff1a;

hive> set hive.input.format;hive.input.format&＃61;org.apache.hadoop.hive.ql.io.HiveInputFormat;

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

简单来说&＃xff0c;InputFormat 主要用于描述输入数据的格式&＃xff0c;提供了以下两个功能&＃xff1a;

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

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

该类接口定义如下&＃xff1a;

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

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

MapReduce 自带了一些 InputFormat 的实现类&＃xff1a;

hive 中有一些 InputFormat 的实现类&＃xff0c;如&＃xff1a;

AvroContainerInputFormatRCFileBlockMergeInputFormatRCFileInputFormatFlatFileInputFormatOneNullRowInputFormatReworkMapredInputFormatSymbolicInputFormatSymlinkTextInputFormatHiveInputFormat

HiveInputFormat 的子类有&＃xff1a;

以 HiveInputFormat 为例&＃xff0c;看看其getSplit()方法逻辑&＃xff1a;

for (Path dir : dirs) {PartitionDesc part &＃61; getPartitionDescFromPath(pathToPartitionInfo, dir);// create a new InputFormat instance if this is the first time to see this// classClass inputFormatClass &＃61; part.getInputFileFormatClass();InputFormat inputFormat &＃61; getInputFormatFromCache(inputFormatClass, job);Utilities.copyTableJobPropertiesToConf(part.getTableDesc(), newjob);// Make filter pushdown information available to getSplits.ArrayList aliases &＃61;mrwork.getPathToAliases().get(dir.toUri().toString());if ((aliases !&＃61; null) && (aliases.size() &＃61;&＃61; 1)) {Operator op &＃61; mrwork.getAliasToWork().get(aliases.get(0));if ((op !&＃61; null) && (op instanceof TableScanOperator)) {TableScanOperator tableScan &＃61; (TableScanOperator) op;pushFilters(newjob, tableScan);}}FileInputFormat.setInputPaths(newjob, dir);newjob.setInputFormat(inputFormat.getClass());InputSplit[] iss &＃61; inputFormat.getSplits(newjob, numSplits / dirs.length);for (InputSplit is : iss) {result.add(new HiveInputSplit(is, inputFormatClass.getName()));}}

上面代码主要过程是&＃xff1a;

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

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

hive> show create table info;OKCREATE 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 DELIMITEDFIELDS TERMINATED BY &＃39;;&＃39;LINES TERMINATED BY &＃39;&＃39;STORED AS INPUTFORMAT&＃39;org.apache.hadoop.mapred.TextInputFormat&＃39;OUTPUTFORMAT&＃39;org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat&＃39;LOCATION&＃39;hdfs://node:8020/user/hive/warehouse/info&＃39;TBLPROPERTIES (&＃39;numPartitions&＃39;&＃61;&＃39;0&＃39;,&＃39;numFiles&＃39;&＃61;&＃39;1&＃39;,&＃39;transient_lastDdlTime&＃39;&＃61;&＃39;1378245263&＃39;,&＃39;numRows&＃39;&＃61;&＃39;0&＃39;,&＃39;totalSize&＃39;&＃61;&＃39;301240320&＃39;,&＃39;rawDataSize&＃39;&＃61;&＃39;0&＃39;)Time taken: 0.497 seconds

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

算法如下&＃xff1a;

long length &＃61; file.getLen();long goalSize &＃61; totalSize / (numSplits &＃61;&＃61; 0 ? 1 : numSplits);long minSize &＃61; Math.max(job.getLong(org.apache.hadoop.mapreduce.lib.input.FileInputFormat.SPLIT_MINSIZE, 1), minSplitSize);long blockSize &＃61; file.getBlockSize();long splitSize &＃61; computeSplitSize(goalSize, minSize, blockSize);long bytesRemaining &＃61; length;while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {String[] splitHosts &＃61; getSplitHosts(blkLocations,length-bytesRemaining, splitSize, clusterMap);splits.add(makeSplit(path, length-bytesRemaining, splitSize,splitHosts));bytesRemaining -&＃61; splitSize;}

华丽的分割线&＃xff1a;以下摘抄自Hadoop深入学习&＃xff1a;InputFormat组件

1)文件切分算法

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

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

这三个参数决定一个 InputFormat 分片的最终的长度&＃xff0c;计算方法如下&＃xff1a;

splitSize &＃61; max{minSize,min{goalSize,blockSize}}

计算出了分片的长度后&＃xff0c;也就确定了 InputFormat 的数目。

2)host 选择算法

InputFormat 的切分方案确定后&＃xff0c;接下来就是要确定每一个 InputSplit 的元数据信息。InputSplit 元数据通常包括四部分&＃xff0c;其意义为&＃xff1a;

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

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

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

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

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

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

getSplits(JobConf job, int numSplits) 代码运行过程如下&＃xff1a;

init(job);CombineFileInputFormatShim combine &＃61; ShimLoader.getHadoopShims().getCombineFileInputFormat();ShimLoader.loadShims(HADOOP_SHIM_CLASSES, HadoopShims.class);Hadoop23ShimsHadoopShimsSecure.getCombineFileInputFormat()

CombineFileInputFormatShim 继承了org.apache.hadoop.mapred.lib.CombineFileInputFormat&＃xff0c;CombineFileInputFormatShim 的 getSplits 方法代码如下&＃xff1a;

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