文档章节

MapReduce剖析笔记之一:从WordCount理解MapReduce的几个阶段

 新司机
发布于 2016/12/30 09:22
字数 4200
阅读 23
收藏 0
WordCount是一个入门的MapReduce程序(从src\examples\org\apache\hadoop\examples粘贴过来的):
package org.apache.hadoop.examples;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

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();
    String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
    if (otherArgs.length != 2) {
      System.err.println("Usage: wordcount <in> <out>");
      System.exit(2);
    }
    Job job = new Job(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(otherArgs[0]));
    FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
    System.exit(job.waitForCompletion(true) ? 0 : 1);
  }
}
MapReduce即将一个计算任务分为两个阶段:Map、Reduce。为什么要这么分解?
为了理解其含义,我们先不管MapReduce这一套框架,从一个简单的问题来看,如果对于100T的日志文件,需要统计其中出现的"ERROR"这个单词的次数,怎么办?
最简单的方法:单机处理,逐行读入每一行文本,统计并累加,则得到其值。问题是:因为数据量太大,速度太慢,怎么办?自然,多机并行处理就是一个自然的选择。
那么,这个文件怎么切分到多个机器呢?假定有100台机器,可以写一个主程序,将这个100T大文件按照每个机器存储1T的原则,在100台机器上分布存储,再把原来单机上的程序拷贝100份(无需修改)至100台机器上运行得到结果,此时得到的结果只是一个中间结果,最后需要写一个汇总程序,将统计结果进行累加,则完成计算。
将大文件分解后,对单机1T文件计算的过程就相当于Map,而Map的结果就相当于"ERROR"这个单词在本机1T文件中出现的次数,而最后的汇总程序就相当于Reduce,Reduce的输入来源于100台机器。在这个简单的例子中,有100个Map任务,1个Reduce任务。
100台机器计算后的中间结果需要传递到Reduce任务所在机器上,这个过程就是Shuffle,Shuffle这个单词的含义是”洗牌“,也就是将中间结果从Map所在机器传输到Reduce所在机器,在这个过程中,存在网络传输。
此时,我们利用上面的例子已经理解了Map-Shuffle-Reduce的基本含义,在上面的例子中,如果还需要对”WARNING“这个单词进行统计,那么怎么办呢?此时,每个Map任务就不仅需要统计本机1T文件中ERROR的个数,还需要统计WARNING的次数,然后在Reduce程序中分别进行统计。如果需要对所有单词进行统计呢?一个道理,每个Map任务对1T文件中所有单词进行统计计数,然后 Reduce对所有结果进行汇总,得到所有单词在100T大文件中出现的次数。此时,问题可能出现了,因为单词数量可能很多,Reduce用单机处理也可能存在瓶颈了,于是我们需要考虑用多台机器并行计算Reduce,假如用2台机器,因为Reduce只是对单词进行计数累加,所有可以按照这样简单的规则进行:大写字母A-Z开头的单词由Reduce 1累加;小写字母a-z开头的单词由Reduce 2累加。
在这种情况下,100个Map任务执行后的结果,都需要分为两部分,一部分准备送到Reduce 1统计,一部分准备送到Reduce 2统计,这个功能称为Partitioner,即将Map后的结果(比如一个文本文件,记录了各个单词在本机文件出现的次数)分解为两部分(比如两个文本 文件),准备送到两个Reduce任务。
因此,Shuffle在这里就是从100个Map任务和2个Reduce任务之间传输中间结果的过程。
我们继续考虑几个问题:
1、 如果Map后的中间结果数据量较大,Shuffle过程对网络带宽要求较高,因此需要将Map后的结果尽可能减小,这个功能当然可以在Map内自己搞 定,不过MapReduce将这个功能单独拎出来,称为Combiner,即合并,这个Combiner,指的是Map任务后中间结果的合并,相比于 Reduce的最终合并,这里相当于先进行一下局部汇总,减小中间结果,进而减小网络传输量。所以,在上面的例子中,假如Map并不计数,只是记录单词出现这个信息,输出结果是<ERROR,1>,<WARNING,1>,<WARNING,1>.....这样一个Key-Value序列,Combiner可以进行局部汇总,将Key相同的Value进行累加,形成一个新的Key-Value序列:<ERROR,14>,<WARNING,27>,.....,这样就大大减小了Shuffle需要的网络带宽,要知道现在数据中心一般使用千兆以太网,好些的使用万兆以太网,TCP/IP传输的效率不太高。这里Combiner汇总函数实际上可以与Reduce的汇总函数一致,只是输入数据不同。
2、 来自100个Map任务后的结果分别送到两个Reduce任务处理。对于任何一个Reduce任务,输入是一堆<ERROR,14>这样的 Key-Value序列,因为100个Map任务都有可能统计到ERROR的次数,因此这里会先进行一个归并,即将相同单词的归并到一起,形 成<ERROR, <14,36,.....>>,<WARNING,<27,45,...>>这样一个仍然是Key-Value的 序列,14、36、。。。分别表示第1、2、。。。台机器中ERROR的统计次数,这个归并过程在MapReduce中称为Merge。如果merge后 再进行Reduce,那么就只需要统计即可;如果事先没有merge,那么Reduce自己完成这一功能也行,只是两种情况下Reduce的输入Key- Value形式不同。
3、如果要求最后的单词统计结果还要形成字典序怎么办呢?可以自己在 Reduce中进行全排序,也可以100个Map任务中分别进行局部排序,然后将结果发到Reduce任务时,再进行归并排序。这个过程 MapReduce也内建支持,因此不需要用户自己去写排序程序,这个过程在MapReduce中称为Sort。
到这里,我们理解了MapReduce中的几个典型步骤:Map、Sort、Partitioner、 Combiner、Shuffle、Merge、Reduce。MapReduce程序之所以称为MapReduce,就说明Map、Reduce这两个 步骤对于一个并行计算来说几乎是必须的,你总得先分开算吧,所以必须有Map;你总得汇总吧,所以有Reduce。当然,理论上也可以不需要 Reduce,如果Map后就得到你要的结果的话。
Sort对于不需要顺序的程序里没意义(但MapReduce默认做了排序);
Partitioner对于Reduce只有一个的时候没意义,如果有多个Reduce,则需要,至于怎么分,用户可以继承Partitioner标准类,自己实现分解函数。控制中间结果如何传输。MapReduce提供的标准的Partitioner是 一个接口,用户可以自己实现getPartition()函数,MapReduce也提供了几个基本的实现,最典型的HashPartitioner是根 据用户设定的Reduce任务数量(注意,MapReduce中,Map任务的个数最终取决于数据分布,Reduce则是用户直接指定),按照哈希进行计算的:
public class HashPartitioner<K2, V2> implements Partitioner<K2, V2> {

  public void configure(JobConf job) {}

  /** Use {@link Object#hashCode()} to partition. */
  public int getPartition(K2 key, V2 value,
                          int numReduceTasks) {
    return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
  }

}
这里,numReduceTasks就是用户设定的Reduce任务数量;
K2 key, V2 value 就是Map计算后的中间结果。
Combiner可以选择性放弃,但考虑到网络带宽,可以自己写相应的函数实现局部合并功能。很多情况下,直接利用Reduce那个程序即可,WordCount这个标准程序里就是这么用的。
Shuffle自然是必须的,不用写,根据Partitioner逻辑,框架自己去执行结果传输。
Merge也不是必须的,可以揉到Reduce里面实现等等也可以。因为这些操作的数据结构都是Key-Value,Reduce的输入只要是一个Key-Value即可,相当灵活。
我们再来看WordCount,这个MapReduce程序中定义了一个类:
public static class TokenizerMapper 
       extends Mapper<Object, Text, Text, IntWritable>
而Mapper是Hadoop中的一个接口,其定义为:
public interface Mapper<K1, V1, K2, V2> extends JobConfigurable, Closeable {
  
  /** 
   * Maps a single input key/value pair into an intermediate key/value pair.
   * 
   * <p>Output pairs need not be of the same types as input pairs.  A given 
   * input pair may map to zero or many output pairs.  Output pairs are 
   * collected with calls to 
   * {@link OutputCollector#collect(Object,Object)}.</p>
   *
   * <p>Applications can use the {@link Reporter} provided to report progress 
   * or just indicate that they are alive. In scenarios where the application 
   * takes an insignificant amount of time to process individual key/value 
   * pairs, this is crucial since the framework might assume that the task has 
   * timed-out and kill that task. The other way of avoiding this is to set 
   * <a href="{@docRoot}/../mapred-default.html#mapred.task.timeout">
   * mapred.task.timeout</a> to a high-enough value (or even zero for no 
   * time-outs).</p>
   * 
   * @param key the input key.
   * @param value the input value.
   * @param output collects mapped keys and values.
   * @param reporter facility to report progress.
   */
  void map(K1 key, V1 value, OutputCollector<K2, V2> output, Reporter reporter)
  throws IOException;
}
因此,Mapper里面并没有规定输入输出的类型是什么,只要是KeyValue的即可,K1、V1、K2、V2是什么由用户指定,反正只是实现K1、V1到K2、V2的映射即可。

在WordCount中实现了继承于Mapper<Object, Text, Text, IntWritable>的一个TokenizerMapper类,实现了map函数:map(Object key, Text value, Context context ) ;

TokenizerMapper中,输入的Key-Value是<Object, Text>,输出是<Text, IntWritable>,在WordCount程序里,K1代表一行文本的起始位置,V1代表这一行文本;

K2代表单词,V2代表"1",用于后面的累和。

同样,在MapReduce中,Reducer也是一个接口,其声明为:
public interface Reducer<K2, V2, K3, V3> extends JobConfigurable, Closeable {
  
  /** 
   * <i>Reduces</i> values for a given key.  
   * 
   * <p>The framework calls this method for each 
   * <code>&lt;key, (list of values)></code> pair in the grouped inputs.
   * Output values must be of the same type as input values.  Input keys must 
   * not be altered. The framework will <b>reuse</b> the key and value objects
   * that are passed into the reduce, therefore the application should clone
   * the objects they want to keep a copy of. In many cases, all values are 
   * combined into zero or one value.
   * </p>
   *   
   * <p>Output pairs are collected with calls to  
   * {@link OutputCollector#collect(Object,Object)}.</p>
   *
   * <p>Applications can use the {@link Reporter} provided to report progress 
   * or just indicate that they are alive. In scenarios where the application 
   * takes an insignificant amount of time to process individual key/value 
   * pairs, this is crucial since the framework might assume that the task has 
   * timed-out and kill that task. The other way of avoiding this is to set 
   * <a href="{@docRoot}/../mapred-default.html#mapred.task.timeout">
   * mapred.task.timeout</a> to a high-enough value (or even zero for no 
   * time-outs).</p>
   * 
   * @param key the key.
   * @param values the list of values to reduce.
   * @param output to collect keys and combined values.
   * @param reporter facility to report progress.
   */
  void reduce(K2 key, Iterator<V2> values,
              OutputCollector<K3, V3> output, Reporter reporter)
    throws IOException;

}
Reducer的输入为K2, V2(这个对应于Mapper输出的经过Shuffle到达Reducer端的K2,V2,), 输出为K3, V3。

在WordCount中,K2为单词,V2为1这个固定值(或者为局部出现次数,取决于是否有Combiner);K3还是单词,V3就是累和值。

而WordCount里存在继承于Reducer<Text, IntWritable, Text, IntWritable>的IntSumReducer类,完成单词计数累加功能。

对于Combiner,实际上MapReduce没有Combiner这个基类(WordCount自然也没有实现),从任务的提交函数来看:
public void setCombinerClass(Class<? extends Reducer> cls
                               ) throws IllegalStateException {
    ensureState(JobState.DEFINE);
    conf.setClass(COMBINE_CLASS_ATTR, cls, Reducer.class);
  }
可以看出,Combiner使用的类实际上符合Reducer。两者是一样的。

再来看作业提交代码:
在之前先说一下Job和Task的区别,一个MapReduce运行流程称为一个Job,中文称“作业”。
在传统的分布式计算领域,一个Job分为多个Task运行。Task中文一般称为任务,在Hadoop中,这种任务有两种:Map和Reduce
所以下面说到Map和Reduce时,指的是任务;说到整个流程时,指的是作业。不过由于疏忽,可能会将作业称为任务的情况。
根据上下文容易区分出来。

 1     Job job = new Job(conf, "word count");
 2     job.setJarByClass(WordCount.class);
 3     job.setMapperClass(TokenizerMapper.class);
 4     job.setCombinerClass(IntSumReducer.class);
 5     job.setReducerClass(IntSumReducer.class);
 6     job.setOutputKeyClass(Text.class);
 7     job.setOutputValueClass(IntWritable.class);
 8     FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
 9     FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
10     System.exit(job.waitForCompletion(true) ? 0 : 1);
第1行创建一个Job对象,Job是MapReduce中提供的一个作业类,其声明为:
public class Job extends JobContext {  
  public static enum JobState {DEFINE, RUNNING};
  private JobState state = JobState.DEFINE;
  private JobClient jobClient;
  private RunningJob info;
.......
之后,设置该作业的运行类,也就是WordCount这个类;

然后设置Map、Combiner、Reduce三个实现类;

之后,设置输出Key和Value的类,这两个类表明了MapReduce作业完毕后的结果。

Key即单词,为一个Text对象,Text是Hadoop提供的一个可以序列化的文本类;

Value为计数,为一个IntWritable对象,IntWritable是Hadoop提供的一个可以序列化的整数类。

之所以不用普通的String和int,是因为输出Key、 Value需要写入HDFS,因此Key和Value都要可写,这种可写能力在Hadoop中使用一个接口Writable表示,其实就相当于序列化,换句话说,Key、Value必须得有可序列化的能力。Writable的声明为:
public interface Writable {
  /** 
   * Serialize the fields of this object to <code>out</code>.
   * 
   * @param out <code>DataOuput</code> to serialize this object into.
   * @throws IOException
   */
  void write(DataOutput out) throws IOException;

  /** 
   * Deserialize the fields of this object from <code>in</code>.  
   * 
   * <p>For efficiency, implementations should attempt to re-use storage in the 
   * existing object where possible.</p>
   * 
   * @param in <code>DataInput</code> to deseriablize this object from.
   * @throws IOException
   */
  void readFields(DataInput in) throws IOException;
}
在第8、9行,还设置了要计算的文件在HDFS中的路径,设定好这些配置和参数后,执行作业提交:job.waitForCompletion(true)
waitForCompletion是Job类中实现的一个方法:
public boolean waitForCompletion(boolean verbose
                                   ) throws IOException, InterruptedException,
                                            ClassNotFoundException {
    if (state == JobState.DEFINE) {
      submit();
    }
    if (verbose) {
      jobClient.monitorAndPrintJob(conf, info);
    } else {
      info.waitForCompletion();
    }
    return isSuccessful();
  }
即执行submit函数:
public void submit() throws IOException, InterruptedException, 
                              ClassNotFoundException {
    ensureState(JobState.DEFINE);
    setUseNewAPI();
    
    // Connect to the JobTracker and submit the job
    connect();
    info = jobClient.submitJobInternal(conf);
    super.setJobID(info.getID());
    state = JobState.RUNNING;
   }
其中,调用jobClient对象的submitJobInternal方法进行作业提交。jobClient是 JobClient对象,在执行connect()的时候即创建出来:
private void connect() throws IOException, InterruptedException {
    ugi.doAs(new PrivilegedExceptionAction<Object>() {
      public Object run() throws IOException {
        jobClient = new JobClient((JobConf) getConfiguration());    
        return null;
      }
    });
  }
创建JobClient的参数是这个作业的配置信息,JobClient是MapReduce作业的客户端部分,主要用于提交作业等等。而具体的作业提交在submitJobInternal方法中实现,关于submitJobInternal的具体实现,包括MapReduce的作业执行流程,较为复杂,留作下一节描述。

关于MapReduce的这一流程,我们也可以看出一些特点:

1、 Map任务之间是不通信的,这与传统的MPI(Message Passing Interface)存在本质区别,这就要求划分后的任务具有独立性。这个要求一方面限制了MapReduce的应用场合,但另一方面对于任务执行出错后的处理十分方便,比如执行某个Map任务的机器挂掉了,可以不管其他Map任务,重新在另一台机器上执行一遍即可。因为底层的数据在HDFS里面,有3 份备份,所以数据冗余搭配上Map的重执行这一能力,可以将集群计算的容错性相比MPI而言大大增强。后续博文会对MPI进行剖析,也会对 MapReduce与传统高性能计算中的并行计算框架进行比较。

2、Map任务的分配与数据的分布关系十分密切,对于上面的例子,这个100T的大文件分布在多台机器上,MapReduce框架会根据文件的实际存储位置分配Map任务,这一过程需要对HDFS有好的理解,在后续博文中会对HDFS中进行剖析。到时候,能更好滴理解MapReduce框架。因为两者是搭配起来使用的。

3、 MapReduce的输入数据来自于HDFS,输出结果也写到HDFS。如果一个事情很复杂,需要分成很多个MapReduce作业反复运行,那么就需要来来回回地从磁盘中搬移数据的过程,速度很慢,后续博文会对Spark这一内存计算框架进行剖析,到时候,能更好滴理解MapReduce性能。

4、MapReduce的输入数据和输出结果也可以来自于HBase,HBase本身搭建于HDFS之上(理论上也可以搭建于其他文件系统),这种应用场合大多需要MapReduce处理一些海量结构化数据。后续博文会对HBase进行剖析。

本文转载自:http://www.cnblogs.com/esingchan/p/3917094.html?utm_source=tuicool&utm_medium=referral

粉丝 0
博文 8
码字总数 596
作品 0
北京
程序员
私信 提问
Hadoop2.X的安装与配置(二)本地模式

在上一篇文章中,我们介绍了Hadoop2.X安装与配置前的准备阶段。 在本地模式配置前,首先完成准备阶段。 点击如下链接,进入准备阶段的配置 https://blog.csdn.net/weixin38187469/article/d...

weixin_38187469
2018/04/16
0
0
【hadoop】16.MapReduce-简介

简介 本章节我们先来了解一些关于MapReduce的理论知识。从本章节您可以学习到:MapReduce的相关知识。 1、概念 Mapreduce是一个分布式运算程序的编程框架,是用户开发“基于hadoop的数据分析...

Areya
01/12
19
0
大数据(hadoop-Mapreduce原理架构)

课程目标: 1:MapReduce的应用场景 2:MapReduce编程模型 3:MapReduce的架构 4:常见MapReduce应用场景 5:总结 MapReduce的定义 源自于Google的MapReduce论文 发表于2004年12月 Hadoop M...

这很耳东先生
04/30
27
0
从分治算法到 MapReduce

从分治算法说起 要说 MapReduce 就不得不说分治算法,而分治算法其实说白了,就是四个字 分而治之 。其实就是将一个复杂的问题分解成多组相同或类似的子问题,对这些子问题再分,然后再分。直...

终日而思一
2018/11/23
38
0
大数据开发 | MapReduce介绍

1. MapReduce 介绍 1.1MapReduce的作用 假设有一个计算文件中单词个数的需求,文件比较多也比较大,在单击运行的时候机器的内存受限,磁盘受限,运算能力受限,而一旦将单机版程序扩展到集群...

嘿你好夏天
2018/04/18
0
0

没有更多内容

加载失败,请刷新页面

加载更多

OSChina 周六乱弹 —— 早上儿子问我他是怎么来的

Osc乱弹歌单(2019)请戳(这里) 【今日歌曲】 @凉小生 :#今日歌曲推荐# 少点戾气,愿你和这个世界温柔以待。中岛美嘉的单曲《僕が死のうと思ったのは (曾经我也想过一了百了)》 《僕が死の...

小小编辑
今天
2.4K
15
Excption与Error包结构,OOM 你遇到过哪些情况,SOF 你遇到过哪些情况

Throwable 是 Java 中所有错误与异常的超类,Throwable 包含两个子类,Error 与 Exception 。用于指示发生了异常情况。 Java 抛出的 Throwable 可以分成三种类型。 被检查异常(checked Exc...

Garphy
今天
41
0
计算机实现原理专题--二进制减法器(二)

在计算机实现原理专题--二进制减法器(一)中说明了基本原理,现准备说明如何来实现。 首先第一步255-b运算相当于对b进行按位取反,因此可将8个非门组成如下图的形式: 由于每次做减法时,我...

FAT_mt
昨天
40
0
好程序员大数据学习路线分享函数+map映射+元祖

好程序员大数据学习路线分享函数+map映射+元祖,大数据各个平台上的语言实现 hadoop 由java实现,2003年至今,三大块:数据处理,数据存储,数据计算 存储: hbase --> 数据成表 处理: hive --> 数...

好程序员官方
昨天
61
0
tabel 中含有复选框的列 数据理解

1、el-ui中实现某一列为复选框 实现多选非常简单: 手动添加一个el-table-column,设type属性为selction即可; 2、@selection-change事件:选项发生勾选状态变化时触发该事件 <el-table @sel...

everthing
昨天
21
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部