在hive on tez引擎下，orc split的计算逻辑

同一条sql，在不同时段执行时生成的mapper数量可能差异差大。为弄明白个中的原因，跟了下源码，大致计算逻辑如下：

mapper数量是和OrcInputFormat中产生的split相关，OrcInputFormat在计算split时会分为两个大的步骤：
1. 计算物理split数量
2. 计算逻辑split数量(对物理split进行分组)

### orc文件的物理结构图
![](/api/file/getImage?fileId=5c063881418f8a54f600022d)
一个orc文件会有1至N个的Stripe、1个file footer、1个postscript组成。每个stripe包含索引(每列的数据在什么位置)、数据、footer(每列分析数据，每列的最大值、最小值等)。footer包含的是文件级别的每列分析数据。postscript是包含orc file的参数，如压缩类型、stripe大小等

### 物理split计算逻辑

* 取出所有相关数据的hdfs文件路径及相关的文件元信息(stripe、blockSize),该操作在`OrcInputFormat.FileGenerator`类中执行
* 基于stripe级别，尝试对同一个文件下的多个stripe打包split。逻辑执行是在`OrcInputFormat.SplitGenerator`类中，以下是关键部分的源码
```java
//current是计算stripes合并时的一个状态类
//context.minSize的值来自于mapreduce.input.fileinputformat.split.minsize
//context.maxSize的值来自于mapreduce.input.fileinputformat.split.maxsize
private OffsetAndLength generateOrUpdateSplit(
        List<OrcSplit> splits, OffsetAndLength current, long offset,
        long length, OrcTail orcTail) throws IOException {
      // if we are working on a stripe, over the min stripe size, and
      // crossed a block boundary, cut the input split here.
      //当current中已存在前面stripe的合并数据，且当于的数据大小大于split的最小值，且数据大小超过一个block的大小时（理论上跨过一个block),将当前的current的状态打包成一个split
      if (current.offset != -1 && current.length > context.minSize &&
          (current.offset / blockSize != offset / blockSize)) {
        splits.add(createSplit(current.offset, current.length, orcTail));
        current.offset = -1;
      }
      // if we aren't building a split, start a new one.
      if (current.offset == -1) {
        current.offset = offset;
        current.length = length;
      } else {
        current.length = (offset + length) - current.offset;
      }
      //当超过上限时(注意允许首次超过上限)，打包split
      if (current.length >= context.maxSize) {
        splits.add(createSplit(current.offset, current.length, orcTail));
        current.offset = -1;
      }
      return current;
    }
```

**划重点：**
1. 一个orc文件至少会生成一个物理的split
2. 当一个orc文件比较大时，可通过控制`mapreduce.input.fileinputformat.split.minsize`、`mapreduce.input.fileinputformat.split.maxsize`来大概控制生成split的数量

### 逻辑split计算逻辑
整个计算逻辑都在`TezSplitGrouper.getGroupedSplits()`方法中，我将其拆分为几个步骤

* 将物理split按照worker主机位置进行整理
```java
Set<String> locSet = new HashSet<String>();
for (SplitContainer split : originalSplits) {
  locSet.clear();
  //取得当前split分布在哪些datanode主机上
  String[] locations = locationProvider.getPreferredLocations(split);
  if (locations == null || locations.length == 0) {
    locations = emptyLocations;
  }
  for (String location : locations) {
    if (location == null) {
      location = emptyLocation;
    }
    locSet.add(location);
  }
  //将分片加入到分布主机的队列中
  for (String location : locSet) {
    LocationHolder holder = distinctLocations.get(location);
    holder.splits.add(split);
  }
}
```
* 重复迭代各主机，让各机拥执行的各分片按照逻辑分片的规则进行分组，确保每轮迭代各主机只获得一个逻辑分片，从而达到数据均衡跑在各主机的目的。逻辑分片的规则具体看代码:
```java
List<SplitContainer> group = new ArrayList<SplitContainer>(numSplitsInGroup);
Set<String> groupLocationSet = new HashSet<String>(10);
for (Map.Entry<String, LocationHolder> entry : distinctLocations.entrySet()) {
        group.clear();
        groupLocationSet.clear();
        String location = entry.getKey(); //datanode主机域名
        LocationHolder holder = entry.getValue();//执行物理分片
        SplitContainer splitContainer = holder.getUnprocessedHeadSplit();//获得未分配的物理分片
        if (splitContainer == null) {
          // all splits on node processed
          continue;
        }
        int oldHeadIndex = holder.headIndex;
        long groupLength = 0;
        int groupNumSplits = 0;
        do {
          group.add(splitContainer);
          groupLength += estimator.getEstimatedSize(splitContainer);
          groupNumSplits++;
          holder.incrementHeadIndex();
          splitContainer = holder.getUnprocessedHeadSplit();
        } while(splitContainer != null
            //若要求按大小分组则分组总大小不超过lengthPerGroup,lengthPerGroup的计算下面给出
            && (!groupByLength || (groupLength + estimator.getEstimatedSize(splitContainer) <= lengthPerGroup)) 
            //若要求按数量分组则物理分片总数不超过numSplitsInGroup,numSplitsInGroup的计算下面给出
            && (!groupByCount || (groupNumSplits + 1 <= numSplitsInGroup)))
            ;

... //略过非关键代码

//创建一个逻辑分片的实例
        GroupedSplitContainer groupedSplit =
            new GroupedSplitContainer(group.size(), wrappedInputFormatName,
                groupLocation,
                // pass rack local hint directly to AM
                ((doingRackLocal && location != emptyLocation)?location:null));
        for (SplitContainer groupedSplitContainer : group) {
          groupedSplit.addSplit(groupedSplitContainer);//将物理分片加入逻辑分片
          Preconditions.checkState(groupedSplitContainer.isProcessed() == false,
              "Duplicates in grouping at location: " + location);
          groupedSplitContainer.setIsProcessed(true);//将物理分片标记为已处理，保证一个物理分片只会分配到一个逻辑分片上
          splitsProcessed++;
        }
        groupedSplits.add(groupedSplit);
      }
```

`lengthPerGroup`和`numSplitsInGroup`的计算过程及其他的一些关键逻辑(以下代码经过整理，没完全按照源码顺序)
```java
    long maxLengthPerGroup = conf.getLong(TEZ_GROUPING_SPLIT_MAX_SIZE,TEZ_GROUPING_SPLIT_MAX_SIZE_DEFAULT);//默认为1G
    long minLengthPerGroup = conf.getLong(TEZ_GROUPING_SPLIT_MIN_SIZE,TEZ_GROUPING_SPLIT_MIN_SIZE_DEFAULT);//默认为50M
    boolean groupByLength = conf.getBoolean(TEZ_GROUPING_SPLIT_BY_LENGTH,TEZ_GROUPING_SPLIT_BY_LENGTH_DEFAULT);//默认为true
    boolean groupByCount = conf.getBoolean(TEZ_GROUPING_SPLIT_BY_COUNT,TEZ_GROUPING_SPLIT_BY_COUNT_DEFAULT);//默认为false
    
    int splitCount = desiredNumSplits>0?desiredNumSplits:originalSplits.size();//desiredNumSplits一般都是1或个位数
    long lengthPerGroup = totalLength/splitCount;//totalLength是所有物理分片的总大小
    
    //当计算的平均分片大小大于配置的最大分片大小时，重新计算desiredNumSplits预期分片数量
    if (lengthPerGroup > maxLengthPerGroup) {
        // splits too big to work. Need to override with max size.
        int newDesiredNumSplits = (int)(totalLength/maxLengthPerGroup) + 1;
        LOG.info("Desired splits: " + desiredNumSplits + " too small. " +
            " Desired splitLength: " + lengthPerGroup +
            " Max splitLength: " + maxLengthPerGroup +
            " New desired splits: " + newDesiredNumSplits +
            " Total length: " + totalLength +
            " Original splits: " + originalSplits.size());

desiredNumSplits = newDesiredNumSplits;
        
      //当计算的平均分片大小小于配置的最大分片大小时，重新计算desiredNumSplits预期分片数量
      } else if (lengthPerGroup < minLengthPerGroup) {
        // splits too small to work. Need to override with size.
        int newDesiredNumSplits = (int)(totalLength/minLengthPerGroup) + 1;
        /**
         * This is a workaround for systems like S3 that pass the same
         * fake hostname for all splits.
         */
        if (!allSplitsHaveLocalhost) {
          desiredNumSplits = newDesiredNumSplits;
        }

LOG.info("Desired splits: " + desiredNumSplits + " too large. " +
            " Desired splitLength: " + lengthPerGroup +
            " Min splitLength: " + minLengthPerGroup +
            " New desired splits: " + newDesiredNumSplits +
            " Final desired splits: " + desiredNumSplits +
            " All splits have localhost: " + allSplitsHaveLocalhost +
            " Total length: " + totalLength +
            " Original splits: " + originalSplits.size());
      }
    
    //这里很重要，当重新计算的desiredNumSplits数量大于物理分片数量时，则不再执行逻辑分片了，直接返回物理分片的结果
    //这里会有些缺陷，当物理分片的大小分布比较极端时，比如有些分片很大，而大部分的分片比较小，当直接返回物理物理分片时，分片效果会不理想
    if (desiredNumSplits == 0 ||
        originalSplits.size() == 0 ||
        desiredNumSplits >= originalSplits.size()) {
      // nothing set. so return all the splits as is
      LOG.info("Using original number of splits: " + originalSplits.size() +
          " desired splits: " + desiredNumSplits);
      groupedSplits = new ArrayList<GroupedSplitContainer>(originalSplits.size());
      for (SplitContainer split : originalSplits) {
        GroupedSplitContainer newSplit =
            new GroupedSplitContainer(1, wrappedInputFormatName, cleanupLocations(locationProvider.getPreferredLocations(split)),
                null);
        newSplit.addSplit(split);
        groupedSplits.add(newSplit);
      }
      return groupedSplits;
    }
    
    long lengthPerGroup = totalLength/desiredNumSplits;//最终的平均分组大小
    int numNodeLocations = distinctLocations.size();
    int numSplitsPerLocation = originalSplits.size()/numNodeLocations;
    int numSplitsInGroup = originalSplits.size()/desiredNumSplits; //最终的平均分组数量
```
    
**划重点：**

* `tez.grouping.min-size` 逻辑分片大小的下限，默认值为50m
* `tez.grouping.max-size` 逻辑分片大小的下限，默认值为1G
* `tez.grouping.by-length`可控制逻辑分片按大小分片，默认值true
* `tez.grouping.by-count`可控制逻辑分片按数量进行分片,默认值false。当为false时且小文件较多时，可能会导致mapper执行时间很长；当为true时，可能会导致mapper数量增多

### 其他
从ApplicationMaster的日志中，也可看到分片的状态信息
```
2018-11-30 03:44:50,399 [INFO] [InputInitializer {Map 1} #0] |grouper.TezSplitGrouper|: Grouping splits in Tez
2018-11-30 03:44:50,436 [INFO] [InputInitializer {Map 1} #0] |grouper.TezSplitGrouper|: Desired splits: 47 too small.  Desired splitLength: 298,556,748,194 Max splitLength: 536,870,912 New desired splits: 26137 Total length: 14,032,167,165,156 Original splits: 25597
2018-11-30 03:44:50,436 [INFO] [InputInitializer {Map 1} #0] |grouper.TezSplitGrouper|: Using original number of splits: 25597 desired splits: 26137

```

```
2018-11-28 02:50:11,361 [INFO] [InputInitializer {Map 1} #0] |grouper.TezSplitGrouper|: Grouping splits in Tez
2018-11-28 02:50:11,413 [INFO] [InputInitializer {Map 1} #0] |grouper.TezSplitGrouper|: Desired splits: 1 too small.  Desired splitLength: 13,715,145,989,311 Max splitLength: 1,073,741,824 New desired splits: 12774 Total length: 13,715,145,989,311 Original splits: 25604
2018-11-28 02:50:11,514 [INFO] [InputInitializer {Map 1} #0] |grouper.TezSplitGrouper|: Desired numSplits: 12774 lengthPerGroup: 1073676686 numLocations: 8 numSplitsPerLocation: 3200 numSplitsInGroup: 2 totalLength: 13715145989311 numOriginalSplits: 25604 . Grouping by length: true count: false nodeLocalOnly: false
2018-11-28 02:50:11,701 [INFO] [InputInitializer {Map 1} #0] |grouper.TezSplitGrouper|: Doing rack local after iteration: 236 splitsProcessed: 19157 numFullGroupsInRound: 0 totalGroups: 1880 lengthPerGroup: 805257472 numSplitsInGroup: 1
2018-11-28 02:50:11,717 [INFO] [InputInitializer {Map 1} #0] |grouper.TezSplitGrouper|: Allowing small groups after iteration: 238 splitsProcessed: 22136 numFullGroupsInRound: 0 totalGroups: 1881
2018-11-28 02:50:11,728 [INFO] [InputInitializer {Map 1} #0] |grouper.TezSplitGrouper|: Number of splits desired: 12774 created: 1885 splitsProcessed: 25604

```

阿川CH