千家信息网

Hadoop2.6.0学习笔记(一)MapReduce介绍

发表于:2025-01-23 作者:千家信息网编辑
千家信息网最后更新 2025年01月23日,鲁春利的工作笔记,谁说程序员不能有文艺范?Hadoop是大数据处理的存储和计算平台,HDFS主要用来实现数据存储,MapReduce实现数据的计算。MapReduce内部已经封装了分布式的计算功能,在
千家信息网最后更新 2025年01月23日Hadoop2.6.0学习笔记(一)MapReduce介绍


鲁春利的工作笔记,谁说程序员不能有文艺范?



Hadoop是大数据处理的存储和计算平台,HDFS主要用来实现数据存储,MapReduce实现数据的计算。

MapReduce内部已经封装了分布式的计算功能,在做业务功能开发时用户只需要继承Mapper和Reducer这两个类,并分别实现map()和reduce()方法即可。

1、Map阶段

读取hdfs中的数据,然后把原始数据进行规范处理,转化为有利于后续进行处理的数据形式。

2、Reduce阶段

接受map阶段输出的数据,自身进行汇总,然后把结果写入到hdfs中。


map和reduce接收的形参是

hadoop1中,jobtracker和tasktracker。

hadoop2中,yarn上有resourcemanager和nodemanager。


Mapper端

# Hadoop提供的Mapper,自定义的Mapper需要继承该类package org.apache.hadoop.mapreduce;public class Mapper {   /**   * Called once at the beginning of the task.   */  protected void setup(Context context) throws IOException, InterruptedException {    // NOTHING  }    /**   * Called once for each key/value pair in the input split.    * Most applications should override this, but the default is the identity function.   */  @SuppressWarnings("unchecked")  protected void map(KEYIN key, VALUEIN value, Context context)   throws IOException, InterruptedException {    context.write((KEYOUT) key, (VALUEOUT) value);  }    /**   * Called once at the end of the task.   */  protected void cleanup(Context context) throws IOException, InterruptedException {    // NOTHING  }     /**   * Expert users can override this method for more complete control over the    *    * @param context   * @throws IOException   */  public void run(Context context) throws IOException, InterruptedException {    setup(context);    try {      while (context.nextKeyValue()) {        map(context.getCurrentKey(), context.getCurrentValue(), context);      }    } finally {      cleanup(context);    }  }}


Reducer

# Hadoop提供的Reducer,自定义的Reducer需要继承该类package org.apache.hadoop.mapreduce;public class Reducer {  /**   * Called once at the start of the task.   */  protected void setup(Context context) throws IOException, InterruptedException {    // NOTHING  }  /**   * This method is called once for each key.    * Most applications will define their reduce class by overriding this method.    * The default implementation is an identity function.   */  @SuppressWarnings("unchecked")  protected void reduce(KEYIN key, Iterable values, Context context)   throws IOException, InterruptedException {    for(VALUEIN value: values) {      context.write((KEYOUT) key, (VALUEOUT) value);    }  }    /**   * Called once at the end of the task.   */  protected void cleanup(Context context) throws IOException, InterruptedException {    // NOTHING  }     /**   * Advanced application writers can use the    * {@link #run(org.apache.hadoop.mapreduce.Reducer.Context)} method to   * control how the reduce task works.   */  public void run(Context context) throws IOException, InterruptedException {    setup(context);    try {      while (context.nextKey()) {        reduce(context.getCurrentKey(), context.getValues(), context);        // If a back up store is used, reset it        Iterator iter = context.getValues().iterator();        if(iter instanceof ReduceContext.ValueIterator) {          ((ReduceContext.ValueIterator)iter).resetBackupStore();                }      }    } finally {      cleanup(context);    }  }}


Map过程

自定义Mapper类继承自该Mapper.class,类Mapper类的四个参数中,前两个为map()函数的输入,后两个为map()函数的输出。

1、读取输入文件内容,解析成形式,每一个对调用一次map()函数;

2、在map()函数中实现自己的业务逻辑,对输入的进行处理,通过上下文对象将处理后的结果以的形式输出;

3、对输出的进行分区;

4、对不同分组的数据,按照key进行排序、分组,相同key的value放到一个集合中;

5、分组后的数据进行归并处理。

说明:

用户指定输入文件的路径,HDFS可以会自动读取文件内容,一般为文本文件(也可以是其他的),每行调用一次map()函数,调用时每行的行偏移量作为key,行内容作为value传入map中;

MR是分布式的计算框架,map与reduce可能都有多个任务在执行,分区的目的是为了确认哪些map输出应该由哪个reduce来进行接收处理。

map端的shuffle过程随着后续的学习再进行补充。

单词计数举例:

[hadoop@nnode hadoop2.6.0]$ hdfs dfs -cat /data/file1.txthello   worldhello   markhuanghello   hadoop[hadoop@nnode hadoop2.6.0]$

每次传入时都是一行行的读取的,每次调用map函数分别传入的数据是<0, hello world>, <12, hello markhuang>, <28, hello hadoop>

在每次map函数处理时,key为LongWritable类型的,无需处理,只需要对接收到的value进行处理即可。由于是需要进行计数,因此需要对value的值进行split,split后每个单词记一次(出现次数1)。

KEYIN, VALUEIN, KEYOUT, VALUEOUT=>IntWritable, Text, Text, IntWritable


Reduce过程

自定义Reducer类继承自Reducer类,类似于Mapper类,并重写reduce方法,实现自己的业务逻辑。

1、对多个map任务的输出,按照不同的分区,通过网络拷贝到不同的reduce节点;

2、对多个任务的输出进行何必、排序,通过自定义业务逻辑进行处理;

3、把reduce的输出保存到指定文件中。

说明:

reduce接收的输入数据Value按key分组(group),而group按照key排序,形成了>的结构。

单词计数举例:

有四组数据, , ,

依次调用reduce方法,并作为key,value传入,在reduce中通过业务逻辑处理。

KEYIN,VALUEIN,KEYOUT,VALUEOUT=>Text,IntWritable, Text,IntWritable

单词计数程序代码:

Map端

package com.lucl.hadoop.mapreduce;import java.io.IOException;import java.util.StringTokenizer;import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Mapper;// map端public class CustomizeMapper extends Mapper {    @Override    protected void map(LongWritable key, Text value, Context context)             throws IOException, InterruptedException {        LongWritable one = new LongWritable(1);        Text word = new Text();                StringTokenizer token = new StringTokenizer(value.toString());        while (token.hasMoreTokens()) {            String v = token.nextToken();            word.set(v);                        context.write(word, one);        }    }}

Reduce端

package com.lucl.hadoop.mapreduce;import java.io.IOException;import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Reducer;// reduce端public class CustomizeReducer extends Reducer {    @Override    protected void reduce(Text key, Iterable values, Context context)            throws IOException, InterruptedException {        int sum = 0;        for (LongWritable intWritable : values) {            sum += intWritable.get();        }        context.write(key, new LongWritable(sum));    }}

驱动类

package com.lucl.hadoop.mapreduce;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.conf.Configured;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;import org.apache.hadoop.util.GenericOptionsParser;import org.apache.hadoop.util.Tool;import org.apache.hadoop.util.ToolRunner;import org.apache.log4j.Logger;/** *  * @author lucl * */public class MyWordCountApp extends Configured implements Tool{    private static final Logger logger = Logger.getLogger(MyWordCountApp.class);        public static void main(String[] args) {        try {            ToolRunner.run(new MyWordCountApp(), args);        } catch (Exception e) {            e.printStackTrace();        }    }    @Override    public int run(String[] args) throws Exception {        Configuration conf = new Configuration();        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();        if (otherArgs.length < 2) {            logger.info("Usage: wordcount  [...] ");            System.exit(2);        }        /**         * 每个map作为一个job任务运行         */        Job job = Job.getInstance(conf , this.getClass().getSimpleName());                job.setJarByClass(MyWordCountApp.class);                /**         * 指定输入文件或目录         */        FileInputFormat.addInputPaths(job, args[0]);    // 目录                /**         * map端相关设置         */        job.setMapperClass(CustomizeMapper.class);        job.setMapOutputKeyClass(Text.class);        job.setMapOutputValueClass(LongWritable.class);                /**         * reduce端相关设置         */        job.setReducerClass(CustomizeReducer.class);        job.setCombinerClass(CustomizeReducer.class);        job.setOutputKeyClass(Text.class);        job.setOutputValueClass(LongWritable.class);                /**         * 指定输出文件目录         */        FileOutputFormat.setOutputPath(job, new Path(args[1]));                return job.waitForCompletion(true) ? 0 : 1;    }}


单词计数程序调用:

[hadoop@nnode code]$ hadoop jar WCApp.jar /data /wc-20151129010115/11/29 00:20:37 INFO client.RMProxy: Connecting to ResourceManager at nnode/192.168.137.117:803215/11/29 00:20:38 INFO input.FileInputFormat: Total input paths to process : 215/11/29 00:20:39 INFO mapreduce.JobSubmitter: number of splits:215/11/29 00:20:39 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1448694510754_000415/11/29 00:20:39 INFO impl.YarnClientImpl: Submitted application application_1448694510754_000415/11/29 00:20:39 INFO mapreduce.Job: The url to track the job: http://nnode:8088/proxy/application_1448694510754_0004/15/11/29 00:20:39 INFO mapreduce.Job: Running job: job_1448694510754_000415/11/29 00:21:10 INFO mapreduce.Job: Job job_1448694510754_0004 running in uber mode : false15/11/29 00:21:10 INFO mapreduce.Job:  map 0% reduce 0/11/29 00:21:41 INFO mapreduce.Job:  map 100% reduce 0/11/29 00:22:01 INFO mapreduce.Job:  map 100% reduce 100/11/29 00:22:02 INFO mapreduce.Job: Job job_1448694510754_0004 completed successfully15/11/29 00:22:02 INFO mapreduce.Job: Counters: 49        File System Counters                FILE: Number of bytes read=134                FILE: Number of bytes written=323865                FILE: Number of read operations=0                FILE: Number of large read operations=0                FILE: Number of write operations=0                HDFS: Number of bytes read=271                HDFS: Number of bytes written=55                HDFS: Number of read operations=9                HDFS: Number of large read operations=0                HDFS: Number of write operations=2        Job Counters                 Launched map tasks=2                Launched reduce tasks=1                Data-local map tasks=2                Total time spent by all maps in occupied slots (ms)=55944                Total time spent by all reduces in occupied slots (ms)=17867                Total time spent by all map tasks (ms)=55944                Total time spent by all reduce tasks (ms)=17867                Total vcore-seconds taken by all map tasks=55944                Total vcore-seconds taken by all reduce tasks=17867                Total megabyte-seconds taken by all map tasks=57286656                Total megabyte-seconds taken by all reduce tasks=18295808        Map-Reduce Framework                Map input records=6                Map output records=12                Map output bytes=170                Map output materialized bytes=140                Input split bytes=188                Combine input records=12                Combine output records=8                Reduce input groups=7                Reduce shuffle bytes=140                Reduce input records=8                Reduce output records=7                Spilled Records=16                Shuffled Maps =2                Failed Shuffles=0                Merged Map outputs=2                GC time elapsed (ms)=315                CPU time spent (ms)=2490                Physical memory (bytes) snapshot=510038016                Virtual memory (bytes) snapshot=2541662208                Total committed heap usage (bytes)=257171456        Shuffle Errors                BAD_ID=0                CONNECTION=0                IO_ERROR=0                WRONG_LENGTH=0                WRONG_MAP=0                WRONG_REDUCE=0        File Input Format Counters                 Bytes Read=83        File Output Format Counters                 Bytes Written=55[hadoop@nnode code]$


单词计数程序输出结果:

[hadoop@nnode ~]$ hdfs dfs -ls /wc-201511290101Found 2 items-rw-r--r--   2 hadoop hadoop          0 2015-11-29 00:22 /wc-201511290101/_SUCCESS-rw-r--r--   2 hadoop hadoop         55 2015-11-29 00:21 /wc-201511290101/part-r-00000[hadoop@nnode ~]$ hdfs dfs -text /wc-201511290101/part-r-000002.3     1fail    1hadoop  4hello   3markhuang       1ok      1world   1[hadoop@nnode ~]$
0