MapReduce编程模型_大数据系统

概述

MR分布式计算框架，应用场景有个共同特点：任务可被分解为相互独立的子问题。
所以MR编程模型的分布式编程方法，5步：

迭代：遍历输入数据，解析为kv对
映射：输入kv对映射为其他kv对
分组：根据key对中间数据进行分组（grouping）
归约：以组为单位对数据进行归约（reduce）
迭代：最终产生的kv对保存到输出文件中

MR API 基本概念

序列化：主要作用两个，永久存储和进程间通信。输入输出数据中的key和value都要是可序列化的，在Hadoop MapReduce中，使一个Java对象可序列化的方法是让其对应的类实现Writable接口，但key是数据排序的关键字，所以还要实现WritableComparable接口。
Reporter参数：是应用程序用来报告完成进度（progress）、设定状态消息（setStatus）以及更新计数器（incrCounter）。
回调机制：常见的设计模式，将工作流内的某个功能按照约定的接口暴露给外部使用者，提供数据或者要求外部使用者提供数据。比如MapReduce给用户暴露了接口Mapper，用户实现了Mapper后，MapReduce运行时环境就会调用它。

InputFormat接口的设计与实现两个功能：

数据切分：将数据切分为多个split，以便确定Map Task个数以及对应的split
为Mapper提供输入数据：给定某个split，将其解析为一个个kv对

public interface InputFormat {

  
  InputSplit[] getSplits(JobConf job, int numSplits) throws IOException;

  
  RecordReader getRecordReader(InputSplit split,
                                     JobConf job, 
                                     Reporter reporter) throws IOException;
}

getSplits方法完成数据切片的功能，将输入数据划分为numSplits个InputSplit，InputSplit两个特点（逻辑分片、可序列化）。
getRecordReader方法返回一个RecordReader对象，该对象将InputSplit解析为若干个kv对，MapReduce框架在Map Task执行过程中，不断调用RecordReader对象中的方法，迭代获取kv对并交给map()函数处理

InputFormat的各种实现

讨论针对文件的InputFormat，实现的基类都是FileInputFormat，派生出TextInputFormat 、KeyValueInputFormat等（如上图）。
基于文件的InputFormat体系的设计思路是：由公共基类FileInputFormat采用统一的方法对各种输入文件进行切分，由各个派生的InputFormat自己提供机制进一步解析InputSplit。

FileInputFormat的实现

最重要的功能就是为各种InputFormat提供统一的getSplits函数，该函数实现中最核心的两个算法是文件切分算法和host选择算法。

文件切分算法
用于确定InputSplit的个数以及每个InputSplit对应的数据段，FileInputFormat以文件为单位切分生成InputSplit，对于每个文件，由三个属性确定其对应的InputSplit个数：

goalSize：根据用户期望的InputSplit计算出来的，即total/numSplits
minSize：InputSplit的最小值，由配置参数mapred.min.split.size确定，默认1
blockSize：文件在hdfs中存储的block大小，默认64MB

这三个参数决定InputSplit的最终大小，计算公式

splitSize确定了，文件被依次切分为splitSize的inputSplit，最后剩余不足一个splitSize的，单独成为一个InputSplit。

host选择算法
这步是确定每个InputSplit的元数据信息，由四部分组成分别表示inputSplit所在的文件、起始位置、长度以及所在的host列表。难点在于host列表如何确定，会直接影响运行过程中的任务本地性。

因为一个大文件对应的block可能遍布整个hadoop集群，而inputsplit的划分算法可能导致一个InputSplit对应多个block，这些block可能位于不同节点上，使得hadoop不可能实现完全的数据本地性，为此，hadoop将数据本地性按照代价划分为三个等级：node locality 、 rack locality 、data center locality，任务调度时，会依次考虑着三个节点的locality , 优先让空闲资源处理本节点上的数据，如果节点上没有可处理的数据，则处理同个机架上的数据，最差情况就是处理别的机架上的数据。

考虑到任务调度的效率，通常不会把所有节点加到inputSplit的host列表中，而是选择包含该inputSplit数据总量最大的前几个节点作为任务调度时判断任务是否具有本地性的主要凭证，为此，FileInputFormat实现了一个启发式算法

首先按照rack包含的数据量对rack排序，然后rack内部按照每个node包含的数据量对node排序，最后取前N个node的host作为inputSplit的host列表，N为block副本数，这样，当任务调度器调度Task时，只要将Task调度给位于host列表的截断，就认为该Task满足本地性。

所以基于这个原理，为了提高Map Task的数据本地性，应尽量使InputSplit大小与block大小相同。

进入FileInputFormat函数里看看

  
  public InputSplit[] getSplits(JobConf job, int numSplits)
    throws IOException {
    StopWatch sw = new StopWatch().start();
    FileStatus[] files = listStatus(job);
    
    // Save the number of input files for metrics/loadgen
    job.setLong(NUM_INPUT_FILES, files.length);
    long totalSize = 0;                           // compute total size
    for (FileStatus file: files) {                // check we have valid files
      if (file.isDirectory()) {
        throw new IOException("Not a file: "+ file.getPath());
      }
      totalSize += 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);

    // generate splits
    ArrayList splits = new ArrayList(numSplits);
    NetworkTopology clusterMap = new NetworkTopology();
    for (FileStatus file: files) {
      Path path = file.getPath();
      long length = file.getLen();
      if (length != 0) {
        FileSystem fs = path.getFileSystem(job);
        BlockLocation[] blkLocations;
        if (file instanceof LocatedFileStatus) {
          blkLocations = ((LocatedFileStatus) file).getBlockLocations();
        } else {
          blkLocations = fs.getFileBlockLocations(file, 0, length);
        }
        if (isSplitable(fs, path)) {
          long blockSize = file.getBlockSize();
          long splitSize = computeSplitSize(goalSize, minSize, blockSize);

          long bytesRemaining = length;
          while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
            String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations,
                length-bytesRemaining, splitSize, clusterMap);
            splits.add(makeSplit(path, length-bytesRemaining, splitSize,
                splitHosts[0], splitHosts[1]));
            bytesRemaining -= splitSize;
          }

          if (bytesRemaining != 0) {
            String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations, length
                - bytesRemaining, bytesRemaining, clusterMap);
            splits.add(makeSplit(path, length - bytesRemaining, bytesRemaining,
                splitHosts[0], splitHosts[1]));
          }
        } else {
          if (LOG.isDebugEnabled()) {
            // Log only if the file is big enough to be splitted
            if (length > Math.min(file.getBlockSize(), minSize)) {
              LOG.debug("File is not splittable so no parallelization "
                  + "is possible: " + file.getPath());
            }
          }
          String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations,0,length,clusterMap);
          splits.add(makeSplit(path, 0, length, splitHosts[0], splitHosts[1]));
        }
      } else { 
        //Create empty hosts array for zero length files
        splits.add(makeSplit(path, 0, length, new String[0]));
      }
    }
    sw.stop();
    if (LOG.isDebugEnabled()) {
      LOG.debug("Total # of splits generated by getSplits: " + splits.size()
          + ", Timetaken: " + sw.now(TimeUnit.MILLISECONDS));
    }
    return splits.toArray(new FileSplit[splits.size()]);
  }

首先进来是获取job的文件信息，FileStatus表示client这边的文件元信息，listStatus(job)返回一个FileStatus对象数组。

@InterfaceAudience.Public
@InterfaceStability.Stable
public class FileStatus implements Writable, Comparable {

  private Path path;
  private long length;
  private boolean isdir;
  private short block_replication;
  private long blocksize;
  private long modification_time;
  private long access_time;
  private FsPermission permission;
  private String owner;
  private String group;
  private Path symlink;
  
  public FileStatus() { this(0, false, 0, 0, 0, 0, null, null, null, null); }
  
  //后面还有很多构造器，没有截进来，可以看到所有属性，都是围绕file的元信息。

然后计算所有文件的总大小totalsize。
计算goalSize: long goalSize = totalSize / (numSplits == 0 ? 1 : numSplits);
计算minSize：long minSize = Math.max(job.getLong(org.apache.hadoop.mapreduce.lib.input.
FileInputFormat.SPLIT_MINSIZE, 1), minSplitSize); 由配置参数确定，默认是1。
然后正式产生分片splits，里面涉及到host选择算法

// generate splits
    ArrayList splits = new ArrayList(numSplits);
    NetworkTopology clusterMap = new NetworkTopology(); //网络拓扑结构
    for (FileStatus file: files) {
      Path path = file.getPath();
      long length = file.getLen();
      if (length != 0) {
        FileSystem fs = path.getFileSystem(job);
        BlockLocation[] blkLocations; 
        //BlockLocation 表示块的网络位置、包含块副本的主机的信息以及其他块元数据（例如，与块关联的文件偏移量、长度、是否损坏等）
        if (file instanceof LocatedFileStatus) {
          blkLocations = ((LocatedFileStatus) file).getBlockLocations();
          //假如这个file就是包含文件块位置的FileStatus类的实例，那就直接调用getBlockLocations函数来获取文件块位置
        } else {
          blkLocations = fs.getFileBlockLocations(file, 0, length);
          //否则就用FileSystem调用getFileBlockLocations来获取文件块位置
        }
        if (isSplitable(fs, path)) {
          long blockSize = file.getBlockSize();  //获取配置参数
          long splitSize = computeSplitSize(goalSize, minSize, blockSize); //计算splitSize.
		
          long bytesRemaining = length;
          while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
          //这应该是TaskTracker上的插槽
          //file长度/splitSize大于slot的话
			
		//后面就进入host选择算法了
		//getSplitHostsAndCachedHosts此函数识别并返回对给定拆分贡献最大的主机。为了计算贡献，机架局部性被视为与主机局部性相同，因此来自贡献最大的机架的主机优先于贡献较少的机架上的主机
		//getSplitHostsAndCachedHosts中会Sort the racks based on their contribution to this split。同时，对rack中还会Sort the hosts in this rack based on their contribution。说白了就是一个启发式算法，先按照rack包含的数据量对rack进行排序，然后在rack内部按照每个node包含的数据量对node排序，
            String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations,
                length-bytesRemaining, splitSize, clusterMap);
            //返回二维数组
            
            splits.add(makeSplit(path, length-bytesRemaining, splitSize,
                splitHosts[0], splitHosts[1]));
            bytesRemaining -= splitSize;
          }
			//如果不能刚好切分
          if (bytesRemaining != 0) {  
            String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations, length
                - bytesRemaining, bytesRemaining, clusterMap);
            splits.add(makeSplit(path, length - bytesRemaining, bytesRemaining,
                splitHosts[0], splitHosts[1]));
          }
        } else {
          if (LOG.isDebugEnabled()) {
            // Log only if the file is big enough to be splitted
            if (length > Math.min(file.getBlockSize(), minSize)) {
              LOG.debug("File is not splittable so no parallelization "
                  + "is possible: " + file.getPath());
            }
          }
          
          String[][] splitHosts = getSplitHostsAndCachedHosts(blkLocations,0,length,clusterMap);
          splits.add(makeSplit(path, 0, length, splitHosts[0], splitHosts[1]));
        }
      } else { 
      //如果length==0
        //Create empty hosts array for zero length files
        splits.add(makeSplit(path, 0, length, new String[0]));
      }
    }
    sw.stop();
    if (LOG.isDebugEnabled()) {
      LOG.debug("Total # of splits generated by getSplits: " + splits.size()
          + ", Timetaken: " + sw.now(TimeUnit.MILLISECONDS));
    }
    return splits.toArray(new FileSplit[splits.size()]);
  }

getRecordReader函数

该函数实现了类似迭代器的功能，将某个InputSplit解析成一个个kv对，具体实现时，考虑两点：

定位记录边界：为了能识别一条完整的记录，记录之间应该添加一些同步标识，对于TextInputFormat，每两条记录之间存在换行符；对于SequenceFileInputFormat，每隔若干条记录会添加固定长度的同步字符串。通过换行符和同步字符产，很容易定位到一个完整记录的起始位置。同时，为了解决记录跨越InputSplit的读取问题，RecordReader规定每个InputSplit的第一条不完整记录划给前一个InputSplit处理。
解析kv对：定位到一条新的记录后，将该纪律分解为key和value，对于TextInputFormat , 每一行的内容即value，该行在文件中的偏移量为key。

举个例子，TextInputFormat的getRecordReader函数

 public RecordReader getRecordReader(
                                          InputSplit genericSplit, JobConf job,
                                          Reporter reporter)
    throws IOException {
    
    reporter.setStatus(genericSplit.toString());
    String delimiter = job.get("textinputformat.record.delimiter");
    byte[] recordDelimiterBytes = null;
    if (null != delimiter) {
      recordDelimiterBytes = delimiter.getBytes(Charsets.UTF_8);
    }
    return new LineRecordReader(job, (FileSplit) genericSplit,
        recordDelimiterBytes);
  }

OutputFormat接口的设计与实现

package org.apache.hadoop.mapred;

import java.io.IOException;

import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.util.Progressable;


@InterfaceAudience.Public
@InterfaceStability.Stable
public interface OutputFormat {

  
  RecordWriter getRecordWriter(FileSystem ignored, JobConf job,
                                     String name, Progressable progress)
  throws IOException;

  
  void checkOutputSpecs(FileSystem ignored, JobConf job) throws IOException;
}

选择FileOutputFormat来分析
基类FileOutputFormat需要提供所有基于文件的OutputFormat实现的公共功能，主要两个：

实现checkOutputSpecs接口：默认功能是检查用户配置的输出目录是否存在，如果存在抛出异常，防止之前的数据被覆盖。
处理side-effect file ：这个file不是最终的输出文件，而是有特殊用途，典型应用是执行推测式任务，在hadoop中，可能有”慢任务“，即拖慢整个作业的执行速度的任务，所以为了优化，hadoop会为之在另外一个节点上启动一个相同的任务，该任务就被称为推测式任务，最先完成任务的计算结果便是这块数据对应的处理结果。所以为了防止两个任务同时往一个输出文件中写入数据时发生冲突，FileOutputFormat会为每个Task的数据创建一个side-effect file，并将产生的数据临时写入该文件，等Task完成后，再移动到最终目录中，具体操作由OutputCommitter完成。

Mapper和Reducer解析

Mapper和Reducer封装了应用程序的数据处理逻辑，为了简化接口，MapReduce要求所有存储在底层分布式文件系统上的数据均要解释成key/value的形式，并交给Mapper/Reducer中的map/reduce函数处理，产生另外一些key/value。
Mapper和Reducer的类体系很类似，分析Mapper。

包括初始化、Map操作、清理三部分

初始化：Mapper继承了JobConfigurable接口，该接口中的configure方法允许通过JobConf参数对Mapper进行初始化
Map操作：MapReduce框架通过InputFormat中的RecordReader从InputSplit获取一个个key/value对，并交给下面的map()函数处理，map()函数的参数除了key,value外，还有OutputCollector和Reporter两个类型的参数，分别用于输出结果和修改Counter值。
清理：Mapper继承Closeable接口，获得close方法，通过实现该方法对Mapper进行清理。

Partitioner接口的设计与实现

Partition作用：对Mapper产生的中间结果进行分区，保证同组数据交给同个Reducer处理，会直接影响Reduce阶段的负载均衡。

它本身包含一个getPartition方法供实现。

 int getPartition(K2 key, V2 value, int numPartitions);

比如HashPatition实现了基于哈希值的分区方法

public class HashPartitioner implements Partitioner {

  public void configure(JobConf job) {}

  
  public int getPartition(K2 key, V2 value,
                          int numReduceTasks) {
    return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
  }

又比如，TotalOrderPartitioner，它用于数据全排序中，在MR环境中，全排序用归并排序，Map阶段局部排序，Reduce全局排序，为了提高全局排序的性能，MapReduce提供了TotalOrderPartitioner，它能够按照大小将数据分为若干个区间（分片），保证后一个区间的所有数据均大于前一个区间数据：
步骤1：数据采样

在Client端通过采样获取分片的分割点，hadoop自带了几个采样算法，如IntercalSampler、RamdomSampler、SplitSampler。
举例：采样数据：b,abc,abd,bcd,abcd,efg,hii,afd,rrr,mnk；经排序后得到：abc,abcd,abd,afd,b,bcd,efg,hii,mnk,rrr；如果Reduce Task个数为4，则采样数据的四等分点为abd,bcd,mnk。

步骤2：Map阶段

两个组件，Mapper和Partioner，其中Mapper可以用IdentityMapper，直接将输入数据输出，Partitioner选TotalOrderPartitioner，将步骤1中获取的分割点把保存到trie树中以便快速定位任意一个记录所在的区间，这样，每个Map Task产生R（reduce task个数）个区间，且区间之间有序。
TotalOrderPartitioner通过trie树（一种多叉有序树）查找每条记录所对应的Reduce Task编号。

步骤3：Reduce阶段

每个Reduce对分配到的区间数据进行局部排序，最终得到全局排序。

MapReduce编程模型

大数据系统相关栏目本月热门文章