Mapper类——hadoop

柔光的暖阳◎ 2022-08-22 04:55 198阅读 0赞

1、Map是一些单个任务。Mapper类就是实现Map任务的类。haddop提供了一个抽象的Mapper基类，程序员需要继承这个基类，并实现其中相关的接口函数

一个示例Mapper类的定义如下：

public static class MyMapper extend Mapper<Object, Text ,Text ,IntWritable>

①Mapper类是Hadoop提供的一个抽象类，程序员可以继承这个基类并实现其中的相关接口函数。它位于 org.apache.hadoop.mapreduce.Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT>;在Mapper中实现的是对大量数据记录或元素的重复处理，并对每个记录或元素做感兴趣的处理、或取感兴趣的中间结果。

②Mapper类中的四个方法

protected void setup（Context context）
    protexted void map(KEYIN key, VALUEIN value, Context context)
    protected void cleanup(Context context)
    public void run(Context context)

其中setup（）方法一般是用于Mapper类实例化时用户程序可能需要做的一些初始化工作（如创建一个全局数据结构，打开一个全局文件，或者建立数据库连接等）；

map（）方法则一般承担主要的处理工作；

cleanup（）方法则是收尾工作，如关闭文件或者执行map（）后的键值对的分发等

③map（）方法

public void map（Object key ， Text value ， Context context） throws IOException，InterruptedException

其中key是传入map的键值，value是对应键值的value值，context是环境对象参数，供程序访问Hadoop的环境对象

map（）方法对输入的键值对进行处理，产生一系列的中间键值对，转换后的中间键值对可以有新的键值类型。输入的键值对可根据实际应用设定，例如文档数据记录可将文本文件中的行或数据表格中的行

Hadoop使用MapReduce框架为每个由作业的InputFormat产生的InputSplit生成一个Map任务。Mapper类可以通过JabContext.getConfiguration()访问作业的配置信息。

public static class TokenizeMapper extends Mapper<Object , Text, Text, IntWritable>{
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();
    //完成词频统计的map方法
        public void map() throws IOException ,InterruptedException{
             StringTokenizer itr = new  StringTokenizer(value.toString());
             while(){
                  word.set(itr.nextToken());
                  context.write(word,one);
               }
        }
    }

④ setup（）和cleanup（）

Mapper类在实例化时将调用一次setup（）方法做一些初始化Mapper类的工作，例如程序需要时可以在setup（）方法中读入一个全局参数，装入一个文件，或者链接一个数据库。

然后系统会为InputSplit中的每一个键值对调用map（）方法，执行程序员编写的计算逻辑。

最后，系统将调用一次cleanup()方法为Mapper类做一些结束清理工作，如关闭在setup（）中打开的文件或建立的数据库连接。而默认情况下，这两个函数什么都不做，除非用户重载实现。

注意： setup（）和cleanup（）仅仅在初始化Mapper实例和Mapper任务结束时由系统作为回调函数分别各做一次，而不是每次调用map（）方法时都去执行一次。

⑤ Mapper输出结果的整理由一个Mapper节点输出的键值对首先会需要进行合并处理，以便key相同的键值对合并为一个键值对。这样做的目的是为了减少大量键值对在网络上传输的开销。系统提供了一个Combiner类来完成这个合并过程。用户还可以定制并制定一个自定义的Combiner，通过JobConf.setCombinerClass(Class)来设置具体所使用的Combiner对象

⑥Mapper输出的中间键值对还需要进行一些整理，以便将中间结果键值对传递给Reduce节点进行后续处理，这个过程也称作Shuffle。这个整理过程中会将key相同的value构成的所有键值对分到同一组。haddop提供了一个Partitioner 类来完成这个分组处理过程。用户可以通过实现一个自定义的Partitioner来控制那些键值对对发送到哪个Reduce节点。

⑦在传送给Reduce节点之前，中间结果键值对还需要按照key值进行排序，以便于后续的处理。这个排序过程将有一个Sort类来完成，用户可以通过JobConf.setOutputKeyConparatorClass(class)来指定的Sort类的比较器，从而控制排序的顺序，但如果是使用的默认的比较器，则不需要进行这个设置。

⑧ Shuffle之后的结果会被分给 各个Reduce节点。简单的说，Combiner是为了减少数据通信开销，中间结果数据进入Reduce节点前进行合并处理，把具有同样主键的数据合并到一起避免重复传送；此外，一个Reduce节点所处理的数据可能会来自多个Map节点，因此，Map节点输出的中间结果需使用一定的策略进行适当的分区（Partitioner）处理，保证相关数据发送到同一个Reduce节点。

⑨实际上Combiner类是在Map节点上执行的，而Partitioner和sort是在Reduce节点上执行的