浅谈MapReduce编程模型

Posted 2022-08-26 fusiji

以Wordcount程序为例，简单描述MapReduce程序的编程模型。

MapReduce程序组成

MapReduce程序一般分成三个部分：

• 一个程序主引导部分；
• 一个Map程序部分；
• 一个Reduce部分。

主引导部分用来设置MapReduce（以下简称 “ MR” ）程序的一些非业务逻辑属性，例如最终生成jar包时指定MR框架执行该程序的入口、MR程序的map和reduce部分是哪个（一个jar包中可能封装了多个程序）、map和reduce部分的输出参数类型、输入数据的来源（路径）和输出数据的路径等。

Map程序部分和Reduce程序部分就是业务逻辑部分，只要按照MR框架要求的格式编写。对于这两个部分，分别有map( )和reduce( )两个关键的函数，这两个函数要么是重写继承其父类Mapper和Reducer的成员函数，要么是实现其所实现接口内定义的具体函数。

map( )和reduce( )方法

Mapper类和Reducer类在继承时要分别提供4个参数，分别代表当前模块的输入、输出数据的类型。在实现map、reduce方法时会用到输入参数（key-value）的类型。而这里的类型并不是Java自身的数据类型，而是hadoop自己的一套数据标准。

在hadoop中，由于要考虑对象序列化，而Java自带的对象序列化机制添加了大量的冗余——少量数据为保证网络传输的正确，冗余的占比已经很多。对于hadoop中的海量数据而言，如果仍采用Java自带的序列化机制，添加的冗余数据的量更大，不利于集群之间频繁且大量的网络通信，因此hadoop有自己的一套序列化标准。

主引导方法

最后用一个job类指定本次作业的其余设置。job中指定：

1. getInstance（）获得job实例。

2. 设置主类即main函数所在jar包的位置，找到这个主函数后，hadoop才能知道到相应的类的依赖关系（即要告诉classload我的jar包中每个程序的主类在哪，紧跟着才能根据主类找到相应的map和reduce类)。

3. 设置map和reduce方法所在的类。

4. 设置map和reduce方法的输出参数的类型。

5. 设置源数据的路径。

6. 设置输出数据的目录地址路径。

7. 设置job提交给集群，waitforcompletion(true)，布尔值表示执行过程是否展示在屏幕上，还是在后台运行。

Wordcount程序示例

package mapreduce.wordcount;

 

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

 

public class Wordcount

 

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException

 

        Configuration conf = new Configuration();

 

        Job job = Job.getInstance(conf);

 

        /*

         *指定jar包中job的起始入口

         */

 

        job.setJarByClass(Wordcount.class);

 

 

        /*

         * 设置map和reduce方法所在的类

         */

 

        job.setMapperClass(WordcountMapper.class);

        job.setReducerClass(WordcountReducer.class);

 

        /*

         * reduce 没有相应的setReduceOutputKey这样的方法。故应先用整体的对map和reduce设置相同的参数

         * 再对map进行输出参数的特定指定

         */

 

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(LongWritable.class);

 

        job.setMapOutputKeyClass(Text.class);

        job.setMapOutputValueClass(LongWritable.class);

 

        /*

         * 输入输出路径设置，当然也可以从main函数参数输入

         */

 

        FileInputFormat.setInputPaths(job, new Path("/wordcount_Test/input"));

        FileOutputFormat.setOutputPath(job, new Path("/wordcount_Test/output2"));

 

        /*

         * 设置是否在shell中显示执行过程

         * 同时该方法会调用启动yarn资源管理框架的方法

         */

        job.waitForCompletion(true);

 

   

 

 

____________________________________________________________________

package mapreduce.wordcount;

 

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.hadoop.util.StringUtils;

 

public class WordcountMapper extends Mapper<LongWritable, Text, Text, LongWritable>

 

    /*

     * 其中key是map程序所读文件的每读一行的文件偏移量，而非业务逻辑上的数据所需的对应key-value

     * 而四个参数的后两个则描述处理后的标准key-value结果，经map程序处理后形成序列化的类型数据，输出到reduce模块。

     * 程序执行时map方法根据数据的特点，可能执行多次

     * 但无论执行多少次。都要等全部输入数据被map处理完，在整体发送给reduce模块

     */

 

    protected void map(LongWritable value, Text key, Context context)

            throws IOException, InterruptedException

 

        //得到文本中的一行

 

        String string = value.toString();

 

        //这里可以用Java自带的分割方法，推荐用Hadoop提供的工具方法

 

        String[] words = StringUtils.split(string, ‘ ‘);

 

        //遍历的方式提交，缓存起来

 

        for(String word:words)

            context.write(new Text(word), new LongWritable(1));

       

 

   

 

 

____________________________________________________________________

package mapreduce.wordcount;

 

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

 

public class WordcountReducer extends Reducer<Text, LongWritable, Text, LongWritable>

 

    /*

     * Reducer后面的四个泛型参数中的前两个必须要和Mapper后面最后两个泛型类型一致

     */

 

    protected void reduce(Text key, Iterable<LongWritable> values,Context context)

            throws IOException, InterruptedException

 

        long count = 0;

 

        for(LongWritable value:values)

            count = count + value.get();

 

       

 

        //reduce的结果仍输出给mapreduce框架

        context.write(key, new LongWritable(count));