Hadoop MapReduce编程 API入门系列之wordcount版本2（六）

 　　这篇博客，给大家，体会不一样的版本编程。

代码

package zhouls.bigdata.myMapReduce.wordcount4;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
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, IntWritable>{

//该方法循环调用，从文件的split中读取每行调用一次，把该行所在的下标为key，该行的内容为value
protected void map(LongWritable key, Text value,
Context context)
throws IOException, InterruptedException {
String[] words = StringUtils.split(value.toString(), ' ');
for(String w :words){
context.write(new Text(w), new IntWritable(1));
}
}
}

package zhouls.bigdata.myMapReduce.wordcount4;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{

//每组调用一次，这一组数据特点：key相同，value可能有多个。
protected void reduce(Text arg0, Iterable<IntWritable> arg1,
Context arg2)
throws IOException, InterruptedException {
int sum =0;
for(IntWritable i: arg1){
sum=sum+i.get();
}
arg2.write(arg0, new IntWritable(sum));
}
}

//System.setProperty("HADOOP_USER_NAME", "root");
//
//1、MR执行环境有两种：本地测试环境，服务器环境
//
//本地测试环境(windows)：(便于调试)
// 在windows的hadoop目录bin目录有一个winutils.exe
// 1、在windows下配置hadoop的环境变量
// 2、拷贝debug工具(winutils.exe)到HADOOP_HOME/bin
// 3、修改hadoop的源码 ，注意：确保项目的lib需要真实安装的jdk的lib
//
// 4、MR调用的代码需要改变：
// a、src不能有服务器的hadoop配置文件(因为，本地是调试，去服务器环境集群那边的)
// b、再调用是使用：
// Configuration config = new Configuration();
// config.set("fs.defaultFS", "hdfs://HadoopMaster:9000");
// config.set("yarn.resourcemanager.hostname", "HadoopMaster");

//服务器环境：(不便于调试)，有两种方式。
//首先需要在src下放置服务器上的hadoop配置文件（都要这一步）
//1、在本地直接调用，执行过程在服务器上（真正企业运行环境）
// a、把MR程序打包（jar），直接放到本地
// b、修改hadoop的源码 ，注意：确保项目的lib需要真实安装的jdk的lib
// c、增加一个属性：
// config.set("mapred.jar", "C:\Users\Administrator\Desktop\wc.jar");
// d、本地执行main方法，servlet调用MR。
//
//
//2、直接在服务器上，使用命令的方式调用，执行过程也在服务器上
// a、把MR程序打包（jar），传送到服务器上
// b、通过： hadoop jar jar路径 类的全限定名
//
//
//
//
//a,1 b,1
//a,3 c,3
//a,2 d,2
//
//
//a,3 c,3
//a,2 d,2
//a,1 b,1
//

package zhouls.bigdata.myMapReduce.wordcount4;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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 RunJob {

public static void main(String[] args) {
Configuration config =new Configuration();
config.set("fs.defaultFS", "hdfs://HadoopMaster:9000");
config.set("yarn.resourcemanager.hostname", "HadoopMaster");
//    config.set("mapred.jar", "C:\Users\Administrator\Desktop\wc.jar");//先打包好wc.jar
try {
FileSystem fs =FileSystem.get(config);

Job job =Job.getInstance(config);
job.setJarByClass(RunJob.class);

job.setJobName("wc");

job.setMapperClass(WordCountMapper.class);
job.setReducerClass(WordCountReducer.class);

job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);

FileInputFormat.addInputPath(job, new Path("/usr/input/wc/wc.txt"));//新建好输入路径，且数据源

Path outpath =new Path("/usr/output/wc");
if(fs.exists(outpath)){
fs.delete(outpath, true);
}
FileOutputFormat.setOutputPath(job, outpath);

boolean f= job.waitForCompletion(true);
if(f){
System.out.println("job任务执行成功");
}
} catch (Exception e) {
e.printStackTrace();
}
}
}