Hadoop(三)MapReduce

矫情吗；* 2021-11-22 07:34 313阅读 0赞

# 一，概述 #

Hadoop MapReduce是一个可以轻松编写应用程序的软件框架，可靠，容错，在大型集群（数千节点）的商用硬件上并行处理大量数据（多TB级别的数据）。

MapReduce是Hadoop的计算核心。

MapReduce通常将输入数据集拆分为独立的块，这些块由Map任务以完全并行的方式处理。框架对Map的输出进行排序，然后输入到reduce任务。通常，作业的输入和输出都存储在文件系统中。该框架负责调度任务，监视任务并重新执行失败的任务。

通常，计算节点和存储节点是相同的，即MapReduce框架和Hadoop分布式文件系统（HDFS）在同一组节点上运行。此配置允许框架有效地在已存在数据的节点上调度任务，从而在集群中产生非常高的聚合带宽。

MapReduce框架由单个主ResourceManager，每个集群节点一个从NodeManager，每个应用程序一个MRAppMaster组成。

最低限度，应用程序通过实现对用的接口或抽象类，来指定输入/输出的位置，并提供map和reduce函数。以上这些，以及作业的其他参数，构成了作业的配置。

接下来，Hadoop作业客户端提交作业（可执行jar）和配置到ResourceManager，ResourceManager负责将作业和配置分发给slave节点，调度任务并监视他们，为作业的客户端提供任务状态，错误信息等响应。

即使Hadoop框架使用Java实现的，MapReduce应用却不需要必须用Java编码。

*  Hadoop Streamming允许用户使用任何可执行文件（如Shell）作为Mapper或者reducer。
 *  Hadoop Pipes是一个SWIG兼容C++的API，用于实现基于C++的MapReduce应用。

# 二，输入和输出 #

MapReduce只在<key,value>键值对上运行，也就是说，MapReduce的输入是一组<key,value>键值对，输出也是一组<key,value>键值对。

key和value必须由框架进行序列化，因此要实现Writable接口，此外，关键类必须实现WritableComparable接口，以便于排序。

MapReduce作业的输入输出类型：

(input) <k1, v1> -> **map** -> <k2, v2> -> **combine** -> <k2, v2> -> **reduce** -> <k3, v3> (output)

# 三，Example：WordCount v1.0 #

通过样例来了解MapReduce的过程，WordCount是一个简单的应用程序，计算给定输入集中每个单词出现的次数，在本地独立，伪分布式或者完全分布式的Hadoop中都可以执行。

## 1，源码 ##

WordCount.java

public class WordCount {
    
      public static class TokenizerMapper
           extends Mapper<Object, Text, Text, IntWritable>{
    
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();
    
        public void map(Object key, Text value, Context context
                        ) throws IOException, InterruptedException {
          StringTokenizer itr = new StringTokenizer(value.toString());
          while (itr.hasMoreTokens()) {
            word.set(itr.nextToken());
            context.write(word, one);
          }
        }
      }
    
      public static class IntSumReducer
           extends Reducer<Text,IntWritable,Text,IntWritable> {
        private IntWritable result = new IntWritable();
    
        public void reduce(Text key, Iterable<IntWritable> values,
                           Context context
                           ) throws IOException, InterruptedException {
          int sum = 0;
          for (IntWritable val : values) {
            sum += val.get();
          }
          result.set(sum);
          context.write(key, result);
        }
      }
    
      public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "word count");
        job.setJarByClass(WordCount.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
      }
    }

## 2，使用 ##

2.1，假设环境变量是这样设置的

export HADOOP_HOME=/home/work/package/hadoop/hadoop-2.7.7/
    export JAVA_HOME=/usr/
    export PATH=$JAVA_HOME/bin:$PATH:$HADOOP_HOME/bin
    export HADOOP_CLASSPATH=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.191.b12-0.el7_5.x86_64/lib/tools.jar

2.2，编译WordCount.java并，创建jar包

\[root@ecs-7bc6-0001 wordcount\]\# hadoop com.sun.tools.javac.Main WordCount.java  
\[root@ecs-7bc6-0001 wordcount\]\# ls  
WordCount.class  WordCount$IntSumReducer.class  WordCount.java  WordCount$TokenizerMapper.class  
\[root@ecs-7bc6-0001 wordcount\]\# jar cf wc.jar WordCount\*.class  
\[root@ecs-7bc6-0001 wordcount\]\# ls  
wc.jar  WordCount.class  WordCount$IntSumReducer.class  WordCount.java  WordCount$TokenizerMapper.class  
2.3，假设hdfs中的两个目录

*  /hadoop/input/----hdfs中的输入目录
 *  /hadoop/output/----hdfs中的输出目录

2.4，输入文件样例

[root@master2 data]# hdfs dfs -copyFromLocal input2 /hadoop/input/input1
    [root@master2 data]# hdfs dfs -copyFromLocal input1 /hadoop/input/input2
    
    [root@master2 data]# hdfs dfs -ls /hadoop/input/              
    Found 2 items
    -rw-r--r--   2 root supergroup         24 2019-07-26 01:40 /hadoop/input/input1
    -rw-r--r--   2 root supergroup         40 2019-07-26 01:40 /hadoop/input/input2
    [root@master2 data]# hdfs dfs -cat /hadoop/input/input1
    Welcome To Hadoop World
    [root@master2 data]# hdfs dfs -cat /hadoop/input/input2
    Hadoop Is Wonderful I Like Hadoop World

2.5，运行WordCount应用

[root@master2 data]# hadoop jar  wc.jar WordCount /hadoop/input /hadoop/output
    [root@master2 data]# hdfs dfs -ls /hadoop/output   
    Found 2 items
    -rw-r--r--   2 root supergroup          0 2019-07-26 01:44 /hadoop/output/_SUCCESS
    -rw-r--r--   2 root supergroup         60 2019-07-26 01:44 /hadoop/output/part-r-00000
    [root@master2 data]# hdfs dfs -cat /hadoop/output/part-r-00000
    Hadoop	3
    I	1
    Is	1
    Like	1
    To	1
    Welcome	1
    Wonderful	1
    World	2

## 运行过程 ##

WordCount应用非常简单。

public void map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {
      StringTokenizer itr = new StringTokenizer(value.toString());
      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);
      }
    }

Mapper实现，通过map方法一次处理一行，由指定的TextInputFormat提供，然后使用StringTokenizer将行拆分为由空格分隔的标记，并生成<<word>,1>的键值对，对于上面的WordCount的示例，第一个文件map后将生成将生成。

< Welcome, 1>
    < To, 1>
    < Hadoop, 1>
    < World, 1>

第二个文件map后生成

< Hadoop, 1>
    < Is, 1>
    < Wonderful, 1>
    < I, 1>
    < Like, 1>
    < Hadoop, 1>
    < World, 1>

后面的章节还会详细了解为给定任务生成的map数量，以及如何以细粒度的方式控制他们。

job.setCombinerClass(IntSumReducer.class);

WordCount也指定一个combiner，combiner在对map生成的一组key进行排序之后，每个map的输出通过本地组合器（与Reducer相同）进行本地聚合。

WordCount第一个map结果combiner之后的结果如下：

< Welcome, 1>
    < To, 1>
    < Hadoop, 1>
    < World, 1>

在第二个map结果combiner之后的结果如下

< Is, 1>
    < Wonderful, 1>
    < I, 1>
    < Like, 1>
    < World, 1>
    < Hadoop, 2>

reduce过程 ，reduce函数乳腺

public void reduce(Text key, Iterable<IntWritable> values,
                       Context context
                       ) throws IOException, InterruptedException {
      int sum = 0;
      for (IntWritable val : values) {
        sum += val.get();
      }
      result.set(sum);
      context.write(key, result);
    }

Reducer的实现，通过reduce方法，指示将key对应的value加起来，这就是每个key出现的次数。这个Job最后的输出是：

< Welcome, 1>
    < To, 1>
    < Is, 1>
    < Wonderful, 1>
    < I, 1>
    < Like, 1>
    < World, 2>
    < Hadoop, 3>

main中指定了Job的各个方面，如输入/输出的路径（从命令行传进来），key/value类型，input/output格式。然后他调用job的waitForCompletion来提交Job并监视Job的执行。

下来要了解有关Job，InputFormat，OutputFormat以及其他接口和类的更多信息。

# 四，用户接口 #

下面介绍如果实现，配置，调用Job。先从Mapper，Reducer接口入手，Job实现这两个接口提供map和reduce方法；然后了解Job，Partitioner，InputFormat，OutputFormat等。最后在了解DistributedCache，IsolationRunner等通用功能。

## Mapper ##

Mapper将输入的key/value键值对映射到一组中间key/value键值对。

Maps是将输入记录转为中间记录的各个task，转换后的中间记录不需要与输入记录类型相同，给定的输入键值对可以映射到0个或多个输出键值对。

Hadoop MapReduce框架为Job的InputFormat生成的每个InputSplit生成一个map任务。

总的来说，“Mapper实现”通过Job.setMapperClass(Class)方法被传递到Job，然后框架为任务的InputSplit中的每个键值对调用map(WritableComparablem Writable, Context)方法。然后，应用程序可以覆写cleanup（Context）方法去执行清理需要的清理逻辑。

输出键值对不需要和输入键值对由相同的类型，给定的输入键值对可以映射到0个或多个输出键值对。通过context.write(WritableComparable, Writable)来输出输出键值对。

应用程序可以使用Counter计算统计信息。

所有一次给定的输出键值对随后被框架分组，传递到Reducer，以确定最终输出。用户可以通过Job.setGroupingComparatorClass(Class)指定Comparator来控制分组。

Mapper输出被排序随后划分到每个Reducer，分区总数与作业的reduce任务数相同。用户可以自定义Partitioner来控制那些键值对转到哪个Reducer。

用户可以选择通过Job.setCombinerClass(Class)指定combiner，以执行中间输出的本地聚合，这样子能够减少从Mapper传递到Reducer的数据量。

中间排序的输出始终以简单（key-len， key， value-len, value）格式存储，应用程序可以控制是否以后如何压缩中间输出，以及通过配置使用CompressionCodeec。

**how many maps？**

map task的数量一般取决于输入的大小，也就是，输入的块数量。map的正确并行度似乎是每个节点大约10-100个map。如果期望10TB的输入数据，且块大小为128MB，你将最终达到82000maps，除非使用Configuration.set(MRJobConfig.NUM\_MAPS,int)来设置它，或者更高。

## Reducer ##

Reducer reduce一组中间值。用户可以使用Job.setNumReduceTask(int)来设置reduce的数量。

总的来说，“Reducer实现”通过Job.setReducerClass(class)方法传递给Job，然后框架分组输入的每个<key, (list of value)>对，然后调用reduce(WritableComparable, Iterable<Writable>, Context)方法。然后，应用程序可以覆写cleanup方法去执行需要的清理。

Reduce由三个主要的阶段：shuffle，sort，reduce

**Shuffle**

Reducer的输入是排过序的Mapper的输出，在这个阶段，框架通过http获取所有mapper的输出的相关分区。

**Sort**

在这个阶段，框架根据keys分组Reducer输入（因为不同的mappers或许会输出相同的key）。Sheffle和Sort阶段并行执行。

**Secondary Sort**

如果中间keys的比较规则与那些在reduce之前的比较规则不同。一种是通过Job.setComparatorClass(Class)指定，另一种通过Job.setGroupingComparatorClass(Class)指定，可以使用这两个API模拟二级排序。

**Reduce**

在这个阶段，每一个分组的输入<key, (list of values)>，都被reduce(WritableComparable Iterable<Writable>, Context)方式调用执行。reduce任务的输出通常通过Context.write(WritableComparable, Writable)方法被写入到文件系统。

应用使用Counter统计数量。

Reducer的输出没有排序。

**How Many Reduces?**

正确的Reduce数量看起来是0.95或者1.75诚意（节点数\*每个节点的最大容器数）。

使用0.95时，所有的reduce都可以立即启动，并在map完成后开始传输map输出。使用1.75时，更快的节点将完成第一轮reduce并启动第二轮reduce，从而更好地实现负载均衡。

增加reduce的数量会增加框架开销，但会增加负载均衡，并降低故障成本。

**Reducer NONE**

如果不需要reduce，将可以将reduce任务的数量设置为0。

这种情况下，Map的输出直接进入FileSystem，使用FileOutputFormat.setOutputPath（Job，Path）设置输出路径，在将map的输出写入FileSystem之前，框架不会对map的输出进行排序。

## Partitioner ##

Partitioner对key空间进行分区，Partitioner控制map中间keys输出的分区，key或者key的子集用于派生分区，通常使用Hash函数，partitions的总数通常和reduce任务的任务保持一致。因此，这控制了m个reduce 任务中的那个中间keys被发送并进行reduce。

HashPartitioner时默认的Partitioner。

## Counter ##

Mapper实现和Reducer实现都可以使用Counter去报告统计数据，Hadoop MapReduce通常是由Mapper，Reducer，Partitioner组成。

## Job配置 ##

Job实质上是MapReduce的job配置。Job是主要的用户接口，用户可以描述一个MapReduce任务给Hadoop框架去执行，框架忠实的按照Job的描述执行Job。

一些参数可能已经被管理员编辑为最终参数，因此无法更改

虽然一些作业参数是直接设置的，如Job.setNumReduceTasks(int)，但是框架或者Job配置其他参数也会互相影响，这样设置起来也比较复杂。

Job通常被用来指定Mapper， Conbiner， Partitioner， Reducer，InputFormat，OutputFormt实现，FileInputFormat指示输入文件集，FileInputFormat.setInputPaths(Job, Path..)或者FileInputFormat.addInputPath(Job,Path)，FileOutputFormat指示输出文件FileOutputFormat.setOutputPath(Path)。

使用Job指定其他高级的方面，例如要使用的Comparator，要放入DistributedCache的文件，是要要锁Job或者中间输出，这么压缩，是否Job任务在*speculative*模式下执行（setMapSpeculativeExecution(boolean)或者etReduceSpeculativeExecution(boolean)），每个任务最大尝试次数（setMaxMapAttempts/或者setMaxReduceAttempts(int)）等。

当然，用户可以所用COnfiguration.set(String, String)或者Configuration.get(String)去设置或者获取应用程序所需的任意参数，但是DistributedCache用于大量制度数据。

## 任务执行和环境 ##

MRAppMaster在单独的jvm进程中将Mapper/Reducer任务作为子进程执行。

子任务继承父MRAppMaster的环境，用户可以通过（mapreduce.\{map|reduce\}.java.opts.parameters）指定额外的选项给子jvm进程，也可以在Job中配置参数，例如，使用-Djava.library.path=<>指定，给run-time链接器去搜索共享库的非标准路径。如果mapreduce.\{map|reduce\}.java.opts参数中包含符号@taskid@，这是用mapreduce任务的taskId值进行替换插值。

下面是一个包含多个参数和替换的示例，显示了jvm gc日志记录，以及无密码启动JVM JMX代理，以便它可以与jconsole查看子内存，线程获取线程dump文件。还设置了map和reduce的最大最小堆内存大小设置为1024M和512M，还未chiild-jvm的java.library.path添加了一条额外的路径。

<property>
      <name>mapreduce.map.java.opts</name>
      <value>
      -Xmx512M -Djava.library.path=/home/mycompany/lib -verbose:gc -Xloggc:/tmp/@taskid@.gc
      -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false
      </value>
    </property>
    
    <property>
      <name>mapreduce.reduce.java.opts</name>
      <value>
      -Xmx1024M -Djava.library.path=/home/mycompany/lib -verbose:gc -Xloggc:/tmp/@taskid@.gc
      -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false
      </value>
    </property>

## 内存管理 ##

用户或者管理员，也可以指定给执行的子任务或者其他紫禁城指定最大虚拟内存，使用mapreduce.\{map|reduce\}.memory.mb.要知道这里设置的值是每个进程的限制，以MB为单位指定，并且该值要大于等于传递给Java虚拟机的-Xmx参数，否则VM可能无法启动。

注意：mapreduce.\{map | reduce\}.java.opts仅用于从MRAppMaster配置已启动的子任务。

框架的某些部分的内存也是可配置的。在map和reduce任务中，通过调整参数影响并发的、读写磁盘数据的频率，来调整框架的性能。监视作业的文件系统计数器 - 特别是相对于从map到reduce的字节计数 - 对于调整这些参数是非常宝贵的。

（等待继续更新）(^\_^)