class KpiWritable implements Writable{
	long upPackNum ;//上行数据包数
	long downPackNum ;//下行数据包数
	long upPayLoad ;//上行总流量
	long downPayLoad ;//下行总流量
	@Override
	public void write(DataOutput out) throws IOException {
		out.writeLong(upPackNum);
		out.writeLong(downPackNum);
		out.writeLong(upPayLoad);
		out.writeLong(downPayLoad);
	}
	//需要注意 按照什么顺序写出去,就按照什么顺序读进来,以为我们的数据写出去之后,是一个流,流是一个一维的.
	//就是从这个方向到那个方向.
	@Override
	public void readFields(DataInput in) throws IOException {
		this.upPackNum = in.readLong();
		this.downPackNum = in.readLong();
		this.upPayLoad = in.readLong();
		this.downPayLoad = in.readLong();
	}
	public KpiWritable() {
	}
	public KpiWritable(long upPackNum, long downPackNum, long upPayLoad,
			long downPayLoad) {
		super();
		set(upPackNum, downPackNum, upPayLoad, downPayLoad);
	}
	public void set(long upPackNum, long downPackNum, long upPayLoad,
			long downPayLoad) {
		this.upPackNum = upPackNum;
		this.downPackNum = downPackNum;
		this.upPayLoad = upPayLoad;
		this.downPayLoad = downPayLoad;
	}
	@Override
	public String toString() {
		return upPackNum + "\t"+downPackNum + "\t"+upPayLoad+"\t"+downPayLoad;
	}
}

　　　　 public static final String INPUT_PATH = "hdfs://hadoop1:9000/kpi";
	public static final String OUT_PATH = "hdfs://hadoop1:9000/kpi_out";
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH),conf);
		if(fileSystem.isDirectory(new Path(OUT_PATH))){
			fileSystem.delete(new Path(OUT_PATH));
		}
		Job job = new Job(conf, KpiApp2.class.getSimpleName());
		job.setJarByClass(KpiApp2.class);//指定代码打包成jar包运行
		FileInputFormat.setInputPaths(job, new Path(INPUT_PATH));//指定数据的输入路径
		//对数据格式化 默认值就是TextInputFormat 可以不用写
		job.setMapperClass(MyMapper.class);//指定自定义map类
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(KpiWritable.class);
		
		//当reduce输出类型与map输出类型一致时,map输出类型可以不设置.使用reduce类型
		job.setReducerClass(MyReducer.class);//指定自定义reducer类
		job.setOutputKeyClass(Text.class);//指定reduce输出key类型
		job.setOutputValueClass(NullWritable.class);//指定reduce输出value类型
		FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));//指定输出路径
		job.waitForCompletion(true);//把作业提交给JobTracker 等待运行结束
		
	}
	/**
	 * 第一个,第二个泛型为LongWritable和Text由解析器TextInputFormat决定的.
	 * 输出类型一个是手机号,value是流量,没办法使用hadoop基本类型表示,只能自己定义类型.
	 */
	public static class MyMapper extends Mapper<LongWritable, Text, Text, KpiWritable>{
		//我们的map对应一行数据,如果我们有1T的数据,我们的Text执行多少次都不可计数,每次都创建一次就特别浪费时间
		//可以把我们的Text放到外边创建.  text对象中的数据不会被复用.
		Text k2 = new Text();
		@Override
		protected void map(LongWritable key, Text value,Context context)
				throws IOException, InterruptedException {
			String line = value.toString();//value就是输入的每一行
			String[] splited = line.split("\t");//制表符分割
			String mobileNumber = splited[1];//手机号
			k2.set(mobileNumber);
			KpiWritable v2 = new KpiWritable(Long.parseLong(splited[6]), Long.parseLong(splited[7]), Long.parseLong(splited[8]), Long.parseLong(splited[9]));
			//我们在写出数据的时候,将map是写出到context中去的,我们在context.write()的时候,已经把数据处理掉了.
			context.write(k2, v2);
		}
	}
	public static class MyReducer extends Reducer<Text, KpiWritable, Text, NullWritable>{
		//经过分组之后,k2到reduce端就是唯一的.
		//v2s中包含的是相同手机号的KpiWritable对象
		//工作中关于mapreduce的算法并不复杂,代码多在数据的清洗.复杂的场景.
		KpiWritable v3 = new KpiWritable();
		@Override
		protected void reduce(Text k2, Iterable<KpiWritable> v2s,Context context)throws IOException, InterruptedException {
			long upPackNum = 0L ;//上行数据包数
			long downPackNum = 0L ;//下行数据包数
			long upPayLoad = 0L ;//上行总流量
			long downPayLoad = 0L ;//下行总流量
			for (KpiWritable kpiWritable : v2s) {
				upPackNum += kpiWritable.upPackNum ;
				downPackNum += kpiWritable.downPackNum ;
				upPayLoad += kpiWritable.upPayLoad ;
				downPayLoad += kpiWritable.downPayLoad ;
			}
			v3.set(upPackNum, downPackNum, upPayLoad, downPayLoad);
			context.write(new Text(k2.toString()+"\t"+upPackNum + "\t"+downPackNum + "\t"+upPayLoad+"\t"+downPayLoad), null);
		}
	}