hadoop(三)--mapreduce程序运行 (原创)

上篇，我们介绍了hadoop中hdfs文件系统，本篇我们继续学习下hadoop的另一大核心：mapreduce。

mapreduce是hadoop中首创的分布式运算框架，它极擅长离线的海量日志数据的分析，离线的批量任务处理。mapreduce将一个任务(job) 划分为两个阶段：map(映射)阶段和reduce(化简)阶段。

map(映射)阶段： 处理单个的、具体的一条一条的数据，可高度并发。
reduce(化简)阶段：对列表的元素进行适当的合并，汇总，得到最终的结果。

写一个小案例： 统计一个(或多个)文件中每个单词出现的次数--wordcount

程序的思路很简单：
map阶段 ， 分析每一个输入文件中每个单词，一行一行第读取(按空格分隔单词)，每读到一个单词计数加1。
reduce阶段，汇总每个单词出现的次数，计数器累加求和。
注：任何基于mapreduce运行模型（框架）编写的用户程序，都必须遵守mapreduce程序的编写规范，即必须含有一个map程序和reduce程序，另外最好还要有个执行mapreduce程序的入口(主函数)程序 。

WCMapper.java       map程序

 
package com.tingcream.hadoopStudy.mapreduce;  
  
  
import java.io.IOException;  
  
import org.apache.commons.lang.StringUtils;  
import org.apache.hadoop.io.LongWritable;  
import org.apache.hadoop.io.Text;  
import org.apache.hadoop.mapreduce.Mapper;  
  
//wordcount  map程序  
//4个泛型中，前两个是指定mapper输入数据的类型，KEYIN是输入的key的类型，VALUEIN是输入的value的类型  
//map 和 reduce 的数据输入输出都是以 key-value对的形式封装的  
//默认情况下，mapreduce框架传递给我们的mapper的输入数据中，key是要处理的文本中一行的起始偏移量，这一行的内容作为value  
public class WCMapper extends Mapper<LongWritable, Text, Text, LongWritable>{  
      
    //mapreduce框架每读一行数据就调用一次该方法  
    @Override  
    protected void map(LongWritable key, Text value,Context context)  
            throws IOException, InterruptedException {  
        //具体业务逻辑就写在这个方法体中，而且我们业务要处理的数据已经被框架传递进来，在方法的参数中 key-value  
        //key 是这一行数据的起始偏移量     value 是这一行的文本内容  
          
        //将这一行的内容转换成string类型  
        String line = value.toString();  
          
        //对这一行的文本按特定分隔符切分  
        String[] words = StringUtils.split(line, " ");  
          
        //遍历这个单词数组输出为kv形式  k：单词    v ： 1  
        for(String word : words){  
              
            context.write(new Text(word), new LongWritable(1));  
              
        }  
    }  
      
}  

WCReducer.java      reduce 程序

package com.tingcream.hadoopStudy.mapreduce;  
  
import java.io.IOException;  
  
import org.apache.hadoop.io.LongWritable;  
import org.apache.hadoop.io.Text;  
import org.apache.hadoop.mapreduce.Reducer;  
  
public class WCReducer extends Reducer<Text, LongWritable, Text, LongWritable>{  
      
      
    //框架在map处理完成之后，将所有kv对缓存起来，进行分组，然后传递一个组<key,valus{}>，调用一次reduce方法  
    //<hello,{1,1,1,1,1,1.....}>  
    @Override  
    protected void reduce(Text key, Iterable<LongWritable> values,Context context)  
            throws IOException, InterruptedException {  
  
        long count = 0;  
        //遍历value的list，进行累加求和  
        for(LongWritable value:values){  
              
            count += value.get();  
        }  
          
        //输出这一个单词的统计结果  
        context.write(key, new LongWritable(count));  
    }  
      
}   

WCRunner.java      runner入口程序

package com.tingcream.hadoopStudy.mapreduce;  
  
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;  
  
/** 
 * hadoop中的一个job作业 
 *  
 * 用来描述一个特定的作业 
 * 比如，该作业使用哪个类作为逻辑处理中的map，哪个作为reduce 
 * 还可以指定该作业要处理的数据所在的路径 
 * 还可以指定改作业输出的结果放到哪个路径 
 * .... 
 *   
 *  mapreduce程序 集群运行模式，提交jar包的yarn集群中运行 
 *  hadoop  jar  xxx.jar     com.xx.cc.XXXRunner  
 */  
public class WCRunner {  
  
    public static void main(String[] args) throws Exception {  
          
        Configuration conf = new Configuration();  
           
          
        Job wcjob = Job.getInstance(conf);  
          
        //设置整个job所用的那些类在哪个jar包  
        wcjob.setJarByClass(WCRunner.class);  
          
          
        //本job使用的mapper和reducer的类  
        wcjob.setMapperClass(WCMapper.class);  
        wcjob.setReducerClass(WCReducer.class);  
          
          
        //指定reduce的输出数据kv类型  
        wcjob.setOutputKeyClass(Text.class);  
        wcjob.setOutputValueClass(LongWritable.class);  
          
        //指定mapper的输出数据kv类型  
        wcjob.setMapOutputKeyClass(Text.class);  
        wcjob.setMapOutputValueClass(LongWritable.class);  
          
          
        //指定要处理的输入数据存放路径  
        FileInputFormat.setInputPaths(wcjob, new Path("hdfs://hadoopNode01:9000/wc/input/")); //hdfs路径  
       
          
        //指定处理结果的输出数据存放路径  
        FileOutputFormat.setOutputPath(wcjob, new Path("hdfs://hadoopNode01:9000/wc/output/"));//hdfs路径  
       
        //将job提交给集群运行   
        wcjob.waitForCompletion(true);  
          
    } 
} 

提交程序到hadoop集群(yarn)中运行，先将项目打jar包(wc.jar)，将jar包上传到hadoop集群任一节点机器(如hadoopNode01）上。
准备测试输入数据     vi  wc-word.txt  内容如下

java is a great program language  
html  is a web design language  
javascript is a great  propram language two  
c is the first program langeuage  

上传wc-word.txt 到hadoop 的hdfs集群中

hadoop  fs -mkdir -p /wc/input
hadoop  fs  -put  wc-word.txt  /wc/input/ 
在hadoop 的yarn集群中运行wc.jar
hadoop   jar   wc.jar   com.tingcream.hadoopStudy.mapreduce.WCRunner       
观察输出结果 
hadoop  fs -cat /wc/output/part-r-00000 

a       3
c       1
design  1
first   1
great   2
html    1
is      4
java    1
javascript      1
langeuage       1
language        3
program 2
propram 1
the     1
two     1
web     1


ok，发现mapreduce计算正确 。

另外，mapreduce程序也可以在本地环境中运行（无须提交jar包到hadoop  yarn集群中运行），本地环境中是无yarn的，相当于一个单节点hadoop服务器，本地运行主要为了快速检验mapreduce程序有无逻辑缺陷、错误，在验证通过后再提交至hadoop集群中运行。由于在本地环境中运行mapreduce程序，数据的输入源一般来自本地文件系统，集群运行模式输入源来自一般来自hdfs文件系统。

WCLocalRunner.java

package com.tingcream.hadoopStudy.mapreduce;  
  
  
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;  
  
  
  
/** 
 * hadoop中的一个job作业 
 *  
 * 用来描述一个特定的作业 
 * 比如，该作业使用哪个类作为逻辑处理中的map，哪个作为reduce 
 * 还可以指定该作业要处理的数据所在的路径 
 * 还可以指定改作业输出的结果放到哪个路径 
 * .... 
 *  
 * mapreduce 本地运行模式，本地不启动集群(yarn集群)，直接在本地跑（单节点） 
 *  右键--》run as --》 java application 即可 
 */  
public class WCLocalRunner {  
  
    public static void main(String[] args) throws Exception {  
          
        Configuration conf = new Configuration();  
           
          
        Job wcjob = Job.getInstance(conf);  
          
        //设置整个job所用的那些类在哪个jar包  
        wcjob.setJarByClass(WCRunner.class);  
          
          
        //本job使用的mapper和reducer的类  
        wcjob.setMapperClass(WCMapper.class);  
        wcjob.setReducerClass(WCReducer.class);  
          
          
        //指定reduce的输出数据kv类型  
        wcjob.setOutputKeyClass(Text.class);  
        wcjob.setOutputValueClass(LongWritable.class);  
          
        //指定mapper的输出数据kv类型  
        wcjob.setMapOutputKeyClass(Text.class);  
        wcjob.setMapOutputValueClass(LongWritable.class);  
          
          
        //指定要处理的输入数据存放路径  
        //FileInputFormat.setInputPaths(wcjob, new Path("hdfs://hadoopNode01:9000/wc/input/")); //hdfs路径  
        FileInputFormat.setInputPaths(wcjob, new Path("f:/12/wc/input/"));//本地路径  
          
        //指定处理结果的输出数据存放路径  
        //FileOutputFormat.setOutputPath(wcjob, new Path("hdfs://hadoopNode01:9000/wc/output/"));//hdfs路径  
        FileOutputFormat.setOutputPath(wcjob, new Path("f:/12/wc/output/"));//本地路径  
          
        //将job提交给集群运行   
        wcjob.waitForCompletion(true);  
 
    }  
}  

在本地环境（本地模式）运行步骤：  主类右键--》Run as   --》java application