Hadoop2.6.0学习笔记（一）MapReduce介绍

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

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

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

读取hdfs中的数据，然后把原始数据进行规范处理，转化为有利于后续进行处理的数据形式。
2、Reduce阶段
接受map阶段输出的数据，自身进行汇总，然后把结果写入到hdfs中。

map和reduce接收的形参是<key,value>。
hadoop1中，jobtracker和tasktracker。

hadoop2中，yarn上有resourcemanager和nodemanager。

Mapper端

# Hadoop提供的Mapper，自定义的Mapper需要继承该类
package org.apache.hadoop.mapreduce;

public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
/**
* 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<KEYIN,VALUEIN,KEYOUT,VALUEOUT> {
/**
* 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<VALUEIN> 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<VALUEIN> iter = context.getValues().iterator();
if(iter instanceof ReduceContext.ValueIterator) {
((ReduceContext.ValueIterator<VALUEIN>)iter).resetBackupStore();
}
}
} finally {
cleanup(context);
}
}
}

Map过程
自定义Mapper类继承自该Mapper.class，类Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>类的四个参数中，前两个为map()函数的输入，后两个为map()函数的输出。
1、读取输入文件内容，解析成<key, value>形式，每一个<key, value>对调用一次map()函数；
2、在map()函数中实现自己的业务逻辑，对输入的<key, value>进行处理，通过上下文对象将处理后的结果以<key, value>的形式输出；
3、对输出的<key, value>进行分区；
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.txt
hello   world
hello   markhuang
hello   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<KEYIN,VALUEIN,KEYOUT,VALUEOUT>类，类似于Mapper类，并重写reduce方法，实现自己的业务逻辑。
1、对多个map任务的输出，按照不同的分区，通过网络拷贝到不同的reduce节点；
2、对多个任务的输出进行何必、排序，通过自定义业务逻辑进行处理；
3、把reduce的输出保存到指定文件中。

说明：
reduce接收的输入数据Value按key分组（group），而group按照key排序，形成了<key, <collection of values>>的结构。
单词计数举例：
有四组数据<hello, {1, 1, 1}>, <world, {1}>, <markhuang, {1}>, <hadoop, {1}>
依次调用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<LongWritable, Text, Text, LongWritable> {
@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<Text, LongWritable, Text, LongWritable> {
@Override
protected void reduce(Text key, Iterable<LongWritable> 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 <in> [<in>...] <out>");
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-201511290101
15/11/29 00:20:37 INFO client.RMProxy: Connecting to ResourceManager at nnode/192.168.137.117:8032
15/11/29 00:20:38 INFO input.FileInputFormat: Total input paths to process : 2
15/11/29 00:20:39 INFO mapreduce.JobSubmitter: number of splits:2
15/11/29 00:20:39 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_1448694510754_0004
15/11/29 00:20:39 INFO impl.YarnClientImpl: Submitted application application_1448694510754_0004
15/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_0004
15/11/29 00:21:10 INFO mapreduce.Job: Job job_1448694510754_0004 running in uber mode : false
15/11/29 00:21:10 INFO mapreduce.Job:  map 0% reduce 0%
15/11/29 00:21:41 INFO mapreduce.Job:  map 100% reduce 0%
15/11/29 00:22:01 INFO mapreduce.Job:  map 100% reduce 100%
15/11/29 00:22:02 INFO mapreduce.Job: Job job_1448694510754_0004 completed successfully
15/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-201511290101
Found 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-00000
2.3     1
fail    1
hadoop  4
hello   3
markhuang       1
ok      1
world   1
[hadoop@nnode ~]$

本文出自 “闷葫芦的世界” 博客，请务必保留此出处http://luchunli.blog.51cto.com/2368057/1676185