进击大数据系列（六）：Hadoop 分布式计算框架 MapReduce

mapreduce.job.ubertask.enable   #是否启用uber模式
mapreduce.job.ubertask.maxmaps   #ubertask的最大map数
mapreduce.job.ubertask.maxreduces #ubertask的最大reduce数
mapreduce.job.ubertask.maxbytes #ubertask最大作业大小

mapreduce.map.memory.mb
mapreduce.map.cpu.vcores
mapreduce.reduce.memory.mb
mapreduce.reduce.cpu.vcores

<property> 
<name>fs.defaultFS</name> 
<value>hdfs://localhost:9000</value> 
</property> 

<property> 
<name>dfs.replication</name> 
<value>1</value> 
</property> 
<property> 
<name>dfs.namenode.name.dir</name> 
<value>D:\software\hadoop\data\namenode</value> 
</property> 
<property> 
<name>dfs.datanode.data.dir</name> 
<value>D:\software\hadoop\data\datanode</value> 
</property> 
<!-- mapred-site.xml--> 
<property> 
<name>mapreduce.framework.name</name> 
<value>yarn</value> 
</property> 

<property> 
<name>yarn.nodemanager.aux-services</name> 
<value>mapreduce_shuffle</value> 
</property> 
<property> 
<name>yarn.nodemanager.aux-services.mapreduce.shuffle.class</name> 
<value>org.apache.hadoop.mapred.ShuffleHandler</value> 
</property> 

set JAVA_HOME=C:\PROGRA~1\Java\jdk1.8.0_271 
set HADOOP_LOG_DIR=%HADOOP_LOG_DIR%\log

单个MapReduce分为：Mapper、Reducer 和 Driver。 也可以多个MapReduce串行执行 
1．Mapper阶段 
（1）用户自定义的Mapper要继承自己的父类 
（2）Mapper的输入数据是KV对的形式（KV的类型可自定义） 
（3）Mapper中的业务逻辑写在map()方法中 
（4）Mapper的输出数据是KV对的形式（KV的类型可自定义） 
（5）map()方法（MapTask进程）对每一个<K,V>调用一次 
2．Reducer阶段 
（2）Reducer的输入数据类型对应Mapper的输出数据类型，也是KV 
（3）Reducer的业务逻辑写在reduce()方法中 
（4）ReduceTask进程对每一组相同k的<k,v>组调用一次reduce()方法 
3．Driver阶段 
相当于YARN集群的客户端，用于提交我们整个程序到YARN集群，提交的是 
封装了MapReduce程序相关运行参数的job对象

<!-- MAVEN包--> 
<dependency> 
<groupId>org.apache.hadoop</groupId> 
<artifactId>hadoop-client</artifactId> 
<version>${hadoop.version}</version> 
</dependency> 
//2.编写好mapreduce代码本地测试(两种常见方式) 
//3.上传jar包到服务器运行 
hadoop jar apache-hadoop-1.0-SNAPSHOT.jar com.wxl.hadoop.mapReduce.wordCount.WordCountDriver /mapreduce/input/word.txt /mapreduce/output

package com.wxl.hadoop.mapReduce.wordCount; 
import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.io.IntWritable; 
import org.apache.hadoop.io.LongWritable; 
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 java.io.IOException; 
import java.util.Date; 
public class WordCountDriver { 
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
///本地测试，正式环境需要注释掉// 
Date date = new Date();//保证输出的目录不重复 
args = new String[]{"D:\\ideawork\\bigdata\\apache-hadoop\\src\\main\\resources\\mapreduce\\input\\word.txt", 
"D:\\ideawork\\bigdata\\apache-hadoop\\src\\main\\resources\\mapreduce\\output\\" + date.getTime()}; 
 
// 1 获取配置信息以及获取 job 对象 
Configuration conf = new Configuration(); 
Job job = Job.getInstance(conf); 
// 2 关联本 Driver 程序的 jar 
job.setJarByClass(WordCountDriver.class); 
// 3 关联 Mapper 和 Reducer 的 jar 
job.setMapperClass(WordCountMapper.class); 
job.setReducerClass(WordCountReducer.class); 
// 4 设置 Mapper 输出的 kv 类型 
job.setMapOutputKeyClass(Text.class); 
job.setMapOutputValueClass(IntWritable.class); 
// 5 设置最终输出 kv 类型 
job.setOutputKeyClass(Text.class); 
job.setOutputValueClass(IntWritable.class); 
// 6 设置输入和输出路径 
FileInputFormat.setInputPaths(job, new Path(args[0])); 
FileOutputFormat.setOutputPath(job, new Path(args[1])); 
// 7 提交 job 
boolean result = job.waitForCompletion(true); 
System.exit(result ? 0 : 1); 
} 
} 
//Map阶段 
class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> { 
Text k = new Text();//分割后的单词 
IntWritable v = new IntWritable(1);//计数 
@Override 
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
// 1 获取一行 
String line = value.toString(); 
// 2 按空格切割 
String[] words = line.split(" "); 
// 3 输出 
for (String word : words) { 
//排除空值 
if (word.trim() == "" || word.length() == 0) { 
continue; 
} 
System.out.println("map输出>>>" + word); 
// 设置输出的key为切割的单词 
k.set(word); 
// 按单词和计数输出 
context.write(k, v); 
} 
} 
} 
//Reducer 
class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> { 
int sum; 
IntWritable v = new IntWritable(); 
@Override 
protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { 
// 1 累加求和 
sum = 0; 
for (IntWritable count : values) { 
sum += count.get(); 
} 
// 2 输出 
v.set(sum); 
// 得到最终的结果 
context.write(key, v); 
} 
}

package com.wxl.hadoop.mapReduce.wordCount; 
import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.conf.Configured; 
import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.io.IntWritable; 
import org.apache.hadoop.io.LongWritable; 
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.Tool; 
import org.apache.hadoop.util.ToolRunner; 
import java.io.IOException; 
import java.util.Date; 
public class WordCountMapReduce extends Configured implements Tool { 
public static void main(String[] args) throws Exception { 
///本地测试，正式环境需要注释掉// 
Date date = new Date();//保证输出的目录不重复 
args = new String[]{"D:\\ideawork\\bigdata\\apache-hadoop\\src\\main\\resources\\mapreduce\\input\\word.txt", 
"D:\\ideawork\\bigdata\\apache-hadoop\\src\\main\\resources\\mapreduce\\output\\" + date.getTime()}; 
 
// run job 
int status = ToolRunner.run(new WordCountMapReduce(), args); 
// exit program 
System.exit(status); 
} 
// Driver 
public int run(String[] args) throws Exception { 
// 1 获取配置信息以及获取 job 对象 
Configuration conf = super.getConf(); 
//设置job名 
Job job = Job.getInstance(conf, this.getClass().getSimpleName()); 
// 2 关联本 Driver 程序的 jar 
job.setJarByClass(this.getClass()); 
// 3 关联 Mapper 和 Reducer 的 jar 
job.setMapperClass(WordCountMapper.class); 
job.setReducerClass(WordCountReducer.class); 
// 4 设置 Mapper 输出的 kv 类型 
job.setMapOutputKeyClass(Text.class); 
job.setMapOutputValueClass(IntWritable.class); 
// 5 设置最终输出 kv 类型 
job.setOutputKeyClass(Text.class); 
job.setOutputValueClass(IntWritable.class); 
// 6 设置输入和输出路径 
FileInputFormat.setInputPaths(job, new Path(args[0])); 
FileOutputFormat.setOutputPath(job, new Path(args[1])); 
// submit job 
boolean status = job.waitForCompletion(true); 
return status ? 0 : 1; 
} 
//Map阶段 
public static class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> { 
Text k = new Text();//分割后的单词 
IntWritable v = new IntWritable(1);//计数 
@Override 
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
// 1 获取一行 
String line = value.toString(); 
// 2 按空格切割 
String[] words = line.split(" "); 
// 3 输出 
for (String word : words) { 
//排除空值 
if (word.trim() == "" || word.length() == 0) { 
continue; 
} 
System.out.println("map输出>>>" + word); 
// 设置输出的key为切割的单词 
k.set(word); 
// 按单词和计数输出 
context.write(k, v); 
} 
} 
} 
//Reducer 
public static class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> { 
int sum; 
IntWritable v = new IntWritable(); 
@Override 
protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { 
// 1 累加求和 
sum = 0; 
for (IntWritable count : values) { 
sum += count.get(); 
} 
// 2 输出 
v.set(sum); 
// 得到最终的结果 
context.write(key, v); 
} 
} 
}