PS: 上图为mapreduce的设计思想，以wordcount为例子，首先把一个任务分我 两个阶段。 第一map阶段负责业务逻辑，第二数据的合并（比如按a-h合并，然后第二个安装i-k）。中间还有还多负责的业务管理逻辑，由mr application master来管理，协调调度。PS: MapReduce程序都是依赖于HDFS，以流的形式读取

yarn
Apache Hadoop YARN （Yet Another Resource Negotiator，另一种资源协调者）是一种新的 Hadoop 资源管理器，它是一个通用资源管理系统，可为上层应用提供统一的资源管理和调度，它的引入为集群在利用率、资源统一管理和数据共享等方面带来了巨大好处。------------------YARN的基本思想是将JobTracker的两个主要功能（资源管理和作业调度/监控）分离，主要方法是创建一个全局的ResourceManager（RM）和若干个针对应用程序的ApplicationMaster（AM）。这里的应用程序是指传统的MapReduce作业或作业的DAG（有向无环图）。

package cn.itcast.bigdata.mr.wcdemo;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;/*** KEYIN: 默认情况下，是mr框架所读到的一行文本的起始偏移量，Long,* 但是在hadoop中有自己的更精简的序列化接口，所以不直接用Long，而用LongWritable* * VALUEIN:默认情况下，是mr框架所读到的一行文本的内容，String，同上，用Text* * KEYOUT：是用户自定义逻辑处理完成之后输出数据中的key，在此处是单词，String，同上，用Text* VALUEOUT：是用户自定义逻辑处理完成之后输出数据中的value，在此处是单词次数，Integer，同上，用IntWritable* * @author**/public class WordcountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{/*** map阶段的业务逻辑就写在自定义的map()方法中* maptask会对每一行输入数据调用一次我们自定义的map()方法*/@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {//将maptask传给我们的文本内容先转换成StringString line = value.toString();//根据空格将这一行切分成单词String[] words = line.split(" ");//将单词输出为<单词，1>for(String word:words){//将单词作为key，将次数1作为value，以便于后续的数据分发，可以根据单词分发，以便于相同单词会到相同的reduce taskcontext.write(new Text(word), new IntWritable(1));}}}

package cn.itcast.bigdata.mr.wcdemo;import java.io.IOException;import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;/*** KEYIN, VALUEIN 对应  mapper输出的KEYOUT,VALUEOUT类型对应* * KEYOUT, VALUEOUT 是自定义reduce逻辑处理结果的输出数据类型* KEYOUT是单词* VLAUEOUT是总次数* @author**/
public class WordcountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{/* 到这里的时候，数据已经按照一定的特性（规律如 统计a-h）  分组好了，所以入参的时候是一个同样的key，values是Itrater专门用来迭代累加；   然后再执行第二组 hello ,1,...指导统计完* <angelababy,1><angelababy,1><angelababy,1><angelababy,1><angelababy,1>* <hello,1><hello,1><hello,1><hello,1><hello,1><hello,1>* <banana,1><banana,1><banana,1><banana,1><banana,1><banana,1>* 入参key，是一组相同单词kv对的key*/@Overrideprotected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {int count=0;/*Iterator<IntWritable> iterator = values.iterator();while(iterator.hasNext()){count += iterator.next().get();}*/for(IntWritable value:values){count += value.get();}context.write(key, new IntWritable(count));}}

package cn.itcast.bigdata.mr.wcdemo;import org.apache.hadoop.conf.Configuration;
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;/*** 相当于一个yarn集群的客户端* 需要在此封装我们的mr程序的相关运行参数，指定jar包* 最后提交给yarn* @author**/
public class WordcountDriver {public static void main(String[] args) throws Exception {if (args == null || args.length == 0) {args = new String[2];args[0] = "hdfs://192.168.8.10:9000/wordcount/input/wordcount.txt";args[1] = "hdfs://192.168.8.10:9000/wordcount/output";}Configuration conf = new Configuration();//设置的没有用!  ??????      因为在linux下允许，所以这些注释了
//        conf.set("HADOOP_USER_NAME", "hadoop");
//        conf.set("dfs.permissions.enabled", "false");// ---因为linux和windows操作系统的环境变量不同，所以不能直接运行在windows上，现在打包到linux系统上，那么，就不用注释了（因为集群上已经配置过了）/*conf.set("mapreduce.framework.name", "yarn");conf.set("yarn.resoucemanager.hostname", "mini1");*/Job job = Job.getInstance(conf);/*job.setJar("/home/hadoop/wc.jar");*///指定本程序的jar包所在的本地路径job.setJarByClass(WordcountDriver.class);//指定本业务job要使用的mapper/Reducer业务类job.setMapperClass(WordcountMapper.class);job.setReducerClass(WordcountReducer.class);//指定mapper输出数据的kv类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);//指定最终输出的数据的kv类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);//指定job的输入原始文件所在目录FileInputFormat.setInputPaths(job, new Path(args[0]));//指定job的输出结果所在目录FileOutputFormat.setOutputPath(job, new Path(args[1]));//将job中配置的相关参数，以及job所用的java类所在的jar包，提交给yarn去运行/*job.submit();*/boolean res = job.waitForCompletion(true);System.exit(res?0:1);}
}

PS:上图应该这么看，首先待处理文件。当submit（）的时候，会看到文件的规模，进行map()划分（启动maptask的多少，这个文件是job.split），然后job.split、wc.jar、job。xml提交到云端的YARN进行管理。然后yarn启动mr appmaster，启动相应的Node Manager进行管理。   执行map（）任务，他是通过InputFormat组件以流的形式读入一定范围的数据进来，每一个任务执行完了，接着通过另一个组件outputController会生成分区且排序的结果文件，再进入reducetask()任务，  每传过来maptask可以根据分区号进行分区，也就是输出的part-r-0000X的变化的值。   在mapreduce线程中，执行业务逻辑，通过outputFormat输出，最后生成文件（读取多个的文件，输出会进行文件）。

PS:  上图为输出的应用结果，   key为单词字符，value是每个单词的个数

1.  PS:start-dfs.sh  //启动应用start-yarn.sh //启动yarn管理器

/*PS：  打印什么内容是 由key   和  输出  value对象的toString方法所决定的。
*/

PS:文件在hdfs上，用jar包进行编译执行。

Ps:由HDFS读取到 日志文件后，根据文件block的大小进行并行度的确定。通常大文件直接有文件块决定，决定时机是在 waitForCompletion这个函数执行时，会生成一个分块的文件。然后再mapreduce执行时，MRAppMaster就可以读取到信息进行调度操作了。

PS:上图为提交数据的完成的信息，里面有包含分块的信息。所有配置参数，这些参数可以让MRAppMaster读取数据

PS :客户端提交ＭＲ程序job的流程

package cn.itcast.bigdata.mr.provinceflow;import java.util.HashMap;import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;/*** K2  V2  对应的是map输出kv的类型，   * @author* 返回的是分区地址的hashcode*/
public class ProvincePartitioner extends Partitioner<Text, FlowBean>{public static HashMap<String, Integer> proviceDict = new HashMap<String, Integer>();static{proviceDict.put("136", 0);proviceDict.put("137", 1);proviceDict.put("138", 2);proviceDict.put("139", 3);}@Overridepublic int getPartition(Text key, FlowBean value, int numPartitions) {String prefix = key.toString().substring(0, 3);Integer provinceId = proviceDict.get(prefix);return provinceId==null?4:provinceId;}
}

package cn.itcast.bigdata.mr.provinceflow;import java.io.IOException;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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;public class FlowCount {static class FlowCountMapper extends Mapper<LongWritable, Text, Text, FlowBean>{@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String line = value.toString();    //将一行内容转成stringString[] fields = line.split("\t");    //切分字段String phoneNbr = fields[1];    //取出手机号long upFlow = Long.parseLong(fields[fields.length-3]);    //取出上行流量下行流量long dFlow = Long.parseLong(fields[fields.length-2]);context.write(new Text(phoneNbr), new FlowBean(upFlow, dFlow));}}static class FlowCountReducer extends Reducer<Text, FlowBean, Text, FlowBean>{//<183323,bean1><183323,bean2><183323,bean3><183323,bean4>.......
        @Overrideprotected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {long sum_upFlow = 0;long sum_dFlow = 0;//遍历所有bean，将其中的上行流量，下行流量分别累加for(FlowBean bean: values){sum_upFlow += bean.getUpFlow();sum_dFlow += bean.getdFlow();}FlowBean resultBean = new FlowBean(sum_upFlow, sum_dFlow);context.write(key, resultBean);}}public static void main(String[] args) throws Exception {Configuration conf = new Configuration();/*conf.set("mapreduce.framework.name", "yarn");conf.set("yarn.resoucemanager.hostname", "mini1");*/Job job = Job.getInstance(conf);/*job.setJar("/home/hadoop/wc.jar");*///指定本程序的jar包所在的本地路径job.setJarByClass(FlowCount.class);//指定本业务job要使用的mapper/Reducer业务类job.setMapperClass(FlowCountMapper.class);job.setReducerClass(FlowCountReducer.class);//指定我们自定义的数据分区器job.setPartitionerClass(ProvincePartitioner.class);//同时指定相应“分区”数量的reducetaskjob.setNumReduceTasks(5);              //数量必须和分区相对应，可以为1  可以大于分区个数。单数少于1到分区数会报错。//指定mapper输出数据的kv类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(FlowBean.class);//指定最终输出的数据的kv类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(FlowBean.class);//指定job的输入原始文件所在目录FileInputFormat.setInputPaths(job, new Path(args[0]));//指定job的输出结果所在目录FileOutputFormat.setOutputPath(job, new Path(args[1]));//将job中配置的相关参数，以及job所用的java类所在的jar包，提交给yarn去运行/*job.submit();*/boolean res = job.waitForCompletion(true);System.exit(res?0:1);}}

package cn.itcast.bigdata.mr.flowsum;import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;import org.apache.hadoop.io.WritableComparable;public class FlowBean implements WritableComparable<FlowBean>{private long upFlow;private long dFlow;private long sumFlow;//反序列化时，需要反射调用空参构造函数，所以要显示定义一个public FlowBean(){}public FlowBean(long upFlow, long dFlow) {this.upFlow = upFlow;this.dFlow = dFlow;this.sumFlow = upFlow + dFlow;}public void set(long upFlow, long dFlow) {this.upFlow = upFlow;this.dFlow = dFlow;this.sumFlow = upFlow + dFlow;}public long getUpFlow() {return upFlow;}public void setUpFlow(long upFlow) {this.upFlow = upFlow;}public long getdFlow() {return dFlow;}public void setdFlow(long dFlow) {this.dFlow = dFlow;}public long getSumFlow() {return sumFlow;}public void setSumFlow(long sumFlow) {this.sumFlow = sumFlow;}/*** 序列化方法*/@Overridepublic void write(DataOutput out) throws IOException {out.writeLong(upFlow);out.writeLong(dFlow);out.writeLong(sumFlow);}/*** 反序列化方法* 注意：反序列化的顺序跟序列化的顺序完全一致*/@Overridepublic void readFields(DataInput in) throws IOException {upFlow = in.readLong();dFlow = in.readLong();sumFlow = in.readLong();}@Overridepublic String toString() {    return upFlow + "\t" + dFlow + "\t" + sumFlow;}@Overridepublic int compareTo(FlowBean o) {return this.sumFlow>o.getSumFlow()?-1:1;    //从大到小, 当前对象和要比较的对象比, 如果当前对象大, 返回-1, 交换他们的位置(自己的理解)
    }}

package cn.itcast.bigdata.mr.flowsum;import java.io.IOException;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import cn.itcast.bigdata.mr.flowsum.FlowCount.FlowCountMapper;
import cn.itcast.bigdata.mr.flowsum.FlowCount.FlowCountReducer;/*** 13480253104 180 180 360 13502468823 7335 110349 117684 13560436666 1116 954* 2070* * @author* */
public class FlowCountSort {static class FlowCountSortMapper extends Mapper<LongWritable, Text, FlowBean, Text> {FlowBean bean = new FlowBean();//！！！虽然这是一个对象，但是是序列化，所以不用担心数据操作Text v = new Text();@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {// 拿到的是上一个统计程序的输出结果，已经是各手机号的总流量信息String line = value.toString();String[] fields = line.split("\t");String phoneNbr = fields[0];long upFlow = Long.parseLong(fields[1]);long dFlow = Long.parseLong(fields[2]);bean.set(upFlow, dFlow);v.set(phoneNbr);context.write(bean, v);}}/*** 根据key来掉, 传过来的是对象, 每个对象都是不一样的, 所以每个对象都调用一次reduce方法* @author: 张政* @date: 2016年4月11日 下午7:08:18* @package_name: day07.sample*/static class FlowCountSortReducer extends Reducer<FlowBean, Text, Text, FlowBean> {// <bean(),phonenbr>
        @Overrideprotected void reduce(FlowBean bean, Iterable<Text> values, Context context) throws IOException, InterruptedException {context.write(values.iterator().next(), bean);}}public static void main(String[] args) throws Exception {Configuration conf = new Configuration();/*conf.set("mapreduce.framework.name", "yarn");conf.set("yarn.resoucemanager.hostname", "mini1");*/Job job = Job.getInstance(conf);/*job.setJar("/home/hadoop/wc.jar");*///指定本程序的jar包所在的本地路径job.setJarByClass(FlowCountSort.class);//指定本业务job要使用的mapper/Reducer业务类job.setMapperClass(FlowCountSortMapper.class);job.setReducerClass(FlowCountSortReducer.class);//指定mapper输出数据的kv类型job.setMapOutputKeyClass(FlowBean.class);job.setMapOutputValueClass(Text.class);//指定最终输出的数据的kv类型job.setOutputKeyClass(Text.class);job.setOutputValueClass(FlowBean.class);//指定job的输入原始文件所在目录FileInputFormat.setInputPaths(job, new Path(args[0]));//指定job的输出结果所在目录
        Path outPath = new Path(args[1]);/*FileSystem fs = FileSystem.get(conf);if(fs.exists(outPath)){fs.delete(outPath, true);}*/FileOutputFormat.setOutputPath(job, outPath);//将job中配置的相关参数，以及job所用的java类所在的jar包，提交给yarn去运行/*job.submit();*/boolean res = job.waitForCompletion(true);System.exit(res?0:1);}}