<|start_filename|>src/main/java/InputOutputFormatTest/MultiInOutput.java<|end_filename|>
package InputOutputFormatTest;

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.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.input.MultipleInputs;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MapFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import ssdut.training.mapreduce.output.MultOutput;
import ssdut.training.mapreduce.output.MultOutput.MultOutputMapper;
import ssdut.training.mapreduce.output.MultOutput.MultOutputReducer;

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

public class MultiInOutput {
	public static class TxtFileMapper extends Mapper<Object, Text, Text, IntWritable> {
		private final static IntWritable one = new IntWritable(1);

		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
			String[] strs = value.toString().split(" ");
			Text date = new Text(strs[0]);
			context.write(date, one);
		}
	}

	public static class CsvFileMapper extends Mapper<Object, Text, Text, IntWritable> {
		private final static IntWritable one = new IntWritable(1);

		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
			String[] strs = value.toString().split(";");//定义csv文件时用了；做分隔符
			context.write(new Text(strs[0]), one);
		}
	}

	public static class MultOutputReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
		// 通过 MultipleOutputs 类控制输出的文件名和输出路径
		// 定义MultipleOutput对象
		private MultipleOutputs<Text, IntWritable> mos;

		// 覆写MultipleOutput对象的setup()初始化和cleanup()关闭mos对象方法
		protected void setup(Context context) {
			mos = new MultipleOutputs<Text, IntWritable>(context);
		}

		protected void cleanup(Context context) throws IOException, InterruptedException {
			mos.close();
		}

		public void reduce(Text key, Iterable<IntWritable> values, Context context)
				throws IOException, InterruptedException {
			int sum = 0;
			for (IntWritable value : values) {
				sum += value.get();
			}
			// 使用MultiOutputs对象替代Context对象输出
			// 1. 输出到不同文件（格式、文件名）
			if (key.toString().startsWith("2015"))
				mos.write("f2015", key, new IntWritable(sum));
			else if (key.toString().startsWith("2016"))
				mos.write("f2016", key, new IntWritable(sum));
			else
				mos.write("f2017", key, new IntWritable(sum));

		}
	}

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		// 1.设置HDFS配置信息
		String namenode_ip = "192.168.17.10";
		String hdfs = "hdfs://" + namenode_ip + ":9000";
		Configuration conf = new Configuration();
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true");

		// 2.设置MapReduce作业配置信息
		String jobName = "MultInputOutput"; // 作业名称
		Job job = Job.getInstance(conf, jobName);
		job.setJarByClass(MultiInOutput.class); // 指定运行时作业类
		job.setJar("export\\MultiInOutput.jar"); // 指定本地jar包
		job.setMapOutputKeyClass(Text.class); // 设置Mapper输出Key类型
		job.setMapOutputValueClass(IntWritable.class); // 设置Mapper输出Value类型
		job.setReducerClass(MultOutputReducer.class); // 指定Reducer类
		// job.setOutputKeyClass(Text.class); //设置Reduce输出Key类型
		// job.setOutputValueClass(IntWritable.class); //设置Reduce输出Value类型
		
		// 3.指定作业多输入路径，及Map所使用的类
		MultipleInputs.addInputPath(job, new Path(hdfs+"/expr/multiinoutput/data/txt"), TextInputFormat.class, TxtFileMapper.class);
		MultipleInputs.addInputPath(job, new Path(hdfs+"/expr/multiinoutput/data/csv"), TextInputFormat.class, CsvFileMapper.class);
		
		// 定义多文件输出的文件名、输出格式、Reduce输出键类型，值类型
		MultipleOutputs.addNamedOutput(job, "f2015", TextOutputFormat.class, Text.class, IntWritable.class);
		MultipleOutputs.addNamedOutput(job, "f2016", SequenceFileOutputFormat.class, Text.class, IntWritable.class);
		MultipleOutputs.addNamedOutput(job, "f2017", MapFileOutputFormat.class, Text.class, IntWritable.class);

		// 设置作业输出路径
		String outputDir = "/expr/multiinoutput/output"; // 实验输出目录
		Path outPath = new Path(hdfs + outputDir);
		FileOutputFormat.setOutputPath(job, outPath);
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(outPath)) {
			fs.delete(outPath, true);
		}

		// 4.运行作业
		System.out.println("Job: " + jobName + " is running...");
		if (job.waitForCompletion(true)) {
			System.out.println("success!");
			System.exit(0);
		} else {
			System.out.println("failed!");
			System.exit(1);
		}
	}

}


<|start_filename|>src/main/java/mapreduceProgram/DateSortAsc.java<|end_filename|>
package mapreduceProgram;

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.IntWritable;
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 DateSortAsc {

	public static class SortMapper extends Mapper<Object, Text, IntWritable, Text> {
		private IntWritable num = new IntWritable();

		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
			String[] strs = value.toString().split("\t");
			num.set(Integer.parseInt(strs[1]));
			// 将次数作为key进行升序排序
			context.write(num, new Text(strs[0]));
			System.out.println(num.get()+","+strs[0]);
		}
	}

	public static class SortReducer extends Reducer<IntWritable, Text, Text, IntWritable> {

		public void reduce(IntWritable key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			for (Text value : values) {
				// 排序后再次颠倒k-v，将日期作为key
				System.out.println(value.toString()+":"+key.get());
				context.write(value, key);
			}
		}
	}

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		// 1.设置HDFS配置信息
		String namenode_ip = "192.168.17.10";
		String hdfs = "hdfs://" + namenode_ip + ":9000";
		Configuration conf = new Configuration();
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true");

		// 2.设置MapReduce作业配置信息
		String jobName = "DateSortAsc"; // 定义作业名称
		Job job = Job.getInstance(conf, jobName);
		job.setJarByClass(DateSortAsc.class); // 指定作业类
		job.setJar("export\\DateSortAsc.jar"); // 指定本地jar包
		
		job.setMapperClass(SortMapper.class); // 指定Mapper类
		job.setMapOutputKeyClass(IntWritable.class); // 设置Mapper输出Key类型
		job.setMapOutputValueClass(Text.class); // 设置Mapper输出Value类型
		
		job.setReducerClass(SortReducer.class); // 指定Reducer类
		job.setOutputKeyClass(Text.class); // 设置Reduce输出Key类型
		job.setOutputValueClass(IntWritable.class); // 设置Reduce输出Value类型

		// 3.设置作业输入和输出路径
		String dataDir = "/workspace/dateSort/data"; // 实验数据目录
		String outputDir = "/workspace/dateSort/output"; // 实验输出目录
		Path inPath = new Path(hdfs + dataDir);
		Path outPath = new Path(hdfs + outputDir);
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outPath);
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(outPath)) {
			fs.delete(outPath, true);
		}

		// 4.运行作业
		System.out.println("Job: " + jobName + " is running...");
		if (job.waitForCompletion(true)) {
			System.out.println("success!");
			System.exit(0);
		} else {
			System.out.println("failed!");
			System.exit(1);
		}
	}

}


<|start_filename|>src/main/java/ssdut/training/mapreduce/minmaxcount/MinMaxCountReducer.java<|end_filename|>
package ssdut.training.mapreduce.minmaxcount;

import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class MinMaxCountReducer extends Reducer<Text, MinMaxCountTuple, Text, MinMaxCountTuple> {
	private MinMaxCountTuple result = new MinMaxCountTuple();

	@Override
	public void reduce(Text key, Iterable<MinMaxCountTuple> values, Context context)
			throws IOException, InterruptedException {
		result.setMin(null);
		result.setMax(null);
		int sum = 0;
		for (MinMaxCountTuple val : values) {
			if (result.getMin() == null || val.getMin().compareTo(result.getMin()) < 0) {
				result.setMin(val.getMin());
			}
			if (result.getMax() == null || val.getMax().compareTo(result.getMax()) > 0) {
				result.setMax(val.getMax());
			}
			sum += val.getCount();
		}
		result.setCount(sum);
		context.write(key, result);
	}
}

<|start_filename|>src/main/java/ssdut/training/mapreduce/peoplerank/PeopleRank.java<|end_filename|>
package ssdut.training.mapreduce.peoplerank;

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;

public class PeopleRank {

	public static class PeopleRankMapper extends Mapper<LongWritable, Text, Text, Text> {
		@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String[] strs = value.toString().split(",");
			context.write(new Text(strs[0]), new Text(strs[1]));
		}
	}

	public static class PeopleRankReducer extends Reducer<Text, Text, Text, Text> {
		@Override
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			StringBuilder sb = new StringBuilder();
			for (Text v : values) {
				sb.append("\t" + v.toString());
			}
			context.write(key, new Text(sb.toString().replaceFirst("\t", "")));	//将开头的制表符去掉
		}
	}
	
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		String namenode_ip = "192.168.17.10";
		String hdfs = "hdfs://" + namenode_ip + ":9000";			
		Configuration conf = new Configuration();		
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true");
		
		String jobName = "PeopleRank";
		Job job = Job.getInstance(conf, jobName);
		job.setJarByClass(PeopleRank.class);
		job.setJar("export\\PeopleRank.jar");
		job.setMapperClass(PeopleRankMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		job.setReducerClass(PeopleRankReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		
		String dataDir = "/expr/peoplerank/data";
		String outputDir = "/expr/peoplerank/output/adjacent";
		Path inPath = new Path(hdfs + dataDir);
		Path outPath = new Path(hdfs + outputDir);
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outPath);		
		FileSystem fs = FileSystem.get(conf);
		if(fs.exists(outPath)) {
			fs.delete(outPath, true);
		}

		System.out.println( "Job: " + jobName + " is running...");
		if(job.waitForCompletion(true)) {
			System.out.println("success!");
			System.exit(0);
		} else {
			System.out.println("failed!");
			System.exit(1);
		}
	}

}


<|start_filename|>src/main/java/InvertedIndex/InvertedJob.java<|end_filename|>
package InvertedIndex;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
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 InvertedJob {
	public static void main(String[] args) throws Exception {		
		String namenode_ip = "192.168.17.10";
		String hdfs = "hdfs://" + namenode_ip + ":9000";			
		Configuration conf = new Configuration();		
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true");
		
		String jobName = "InvertedJob";
		Job job = Job.getInstance(conf, jobName);
		job.setJarByClass(InvertedJob.class);
		job.setJar("export\\InvertedJob.jar");
		job.setMapperClass(InvertedMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		job.setCombinerClass(InvertedCombiner.class);	//此处定义Combiner类，与Reducer类不同
		job.setReducerClass(InvertedReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		
		String dataDir = "/expr/inverted/data";	
		String outputDir = "/expr/inverted/output";
		Path inPath = new Path(hdfs + dataDir);
		Path outPath = new Path(hdfs + outputDir);
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outPath);		
		FileSystem fs = FileSystem.get(conf);
		if(fs.exists(outPath)) {
			fs.delete(outPath, true);
		}

		System.out.println("Job: " + jobName + " is running...");
		if(job.waitForCompletion(true)) {
			System.out.println("success!");
			System.exit(0);
		} else {
			System.out.println("failed!");
			System.exit(1);
		}		
	}
	
}


<|start_filename|>src/main/java/InvertedIndex/InvertedMapper.java<|end_filename|>
package InvertedIndex;

import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

public class InvertedMapper extends Mapper<Object, Text, Text, Text> {
	private Text keyInfo = new Text();
	private Text valueInfo = new Text();
	private FileSplit split;

	public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
		split = (FileSplit) context.getInputSplit();	//通过context获取输入分片对象，目的是获得输入文件名称
		StringTokenizer itr = new StringTokenizer(value.toString());
		while (itr.hasMoreTokens()) {
			keyInfo.set(itr.nextToken() + ":" + split.getPath().getName()); //将单词及其所属文件拼接成"word:filename"格式作为key
			valueInfo.set("1");
			context.write(keyInfo, valueInfo);  //输出格式： "word:filename	1"
		}
	}
}


<|start_filename|>src/main/java/mapreduceProgram/FlowStatistics.java<|end_filename|>
package mapreduceProgram;

import java.io.DataInput;
import java.io.DataOutput;
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.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Job;

public class FlowStatistics {

	public static class FlowMapper extends Mapper<Object, Text, Text, Text>{
		
		public void map(Object key,Text value,Context context) throws IOException, InterruptedException {
			String[] strs = value.toString().split("\t");
			Text phone = new Text(strs[0]);
			Text flow = new Text(strs[1]+"\t"+strs[2]);
			context.write(phone, flow);
		}
	}
	
	public static class FlowReducer extends Reducer<Text, Text, Text, Text>{
		public void reduce(Text key,Iterable<Text> values,Context context) throws IOException, InterruptedException {
			int upFlow = 0;
			int downFlow = 0;
			
			for (Text value : values) {
				String[] strs = value.toString().split("\t");
				upFlow += Integer.parseInt(strs[0].toString());
				downFlow += Integer.parseInt(strs[1].toString());
			}
			int sumFlow = upFlow+downFlow;
			
			context.write(key,new Text(upFlow+"\t"+downFlow+"\t"+sumFlow));
		}
	}
	
	// 第二种写法
	public static class FlowWritableMapper extends Mapper<Object, Text, Text, FlowWritable> {
		public void map(Object key,Text value,Context context) throws IOException, InterruptedException {
			String[] strs = value.toString().split("\t");
			Text phone = new Text(strs[0]);
			FlowWritable flow = new FlowWritable(Integer.parseInt(strs[1]),Integer.parseInt(strs[2]));
			System.out.println("Flow is:"+flow.toString());
			context.write(phone, flow);
		}
	}
	public static class FlowWritableReducer extends Reducer<Text, FlowWritable, Text, FlowWritable>{
		public void reduce(Text key,Iterable<FlowWritable> values,Context context) throws IOException, InterruptedException {
			int upFlow = 0;
			int downFlow = 0;
			
			for (FlowWritable value : values) {
				upFlow += value.getUpFlow();
				downFlow += value.getDownFlow();
			}
			System.out.println(key.toString()+":"+upFlow+","+downFlow);
			context.write(key,new FlowWritable(upFlow,downFlow));
		}
	}
	
	public static class FlowWritable implements Writable{
		private int upFlow;
		private int downFlow;
		private int sumFlow;
		
		public FlowWritable() {}

		public FlowWritable(int upFlow,int downFlow) {
			this.upFlow = upFlow;
			this.downFlow = downFlow;
			this.sumFlow = upFlow+downFlow;
		}
		
		public int getDownFlow() {
			return downFlow;
		}

		public void setDownFlow(int downFlow) {
			this.downFlow = downFlow;
		}

		public int getUpFlow() {
			return upFlow;
		}

		public void setUpFlow(int upFlow) {
			this.upFlow = upFlow;
		}

		public int getSumFlow() {
			return sumFlow;
		}

		public void setSumFlow(int sumFlow) {
			this.sumFlow = sumFlow;
		}
		
		@Override
		public void write(DataOutput out) throws IOException {
			// TODO Auto-generated method stub
			out.writeInt(upFlow);
			out.writeInt(downFlow);
			out.writeInt(sumFlow);
		}

		@Override
		public void readFields(DataInput in) throws IOException {
			// TODO Auto-generated method stub
			upFlow = in.readInt();
			downFlow = in.readInt();
			sumFlow = in.readInt();
		}

		@Override
		public String toString() {
			// TODO Auto-generated method stub
			return upFlow+"\t"+downFlow+"\t"+sumFlow;
		}
	}
	
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		// 设置hdfs配置信息
		String namenode_ip = "192.168.17.10";
		String hdfs = "hdfs://"+namenode_ip+":9000";
		Configuration conf = new Configuration();
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true");
		
		// 设置作业Job配置信息
		String jobName = "FlowStatistics";
		Job job = Job.getInstance(conf, jobName);
		job.setJarByClass(FlowStatistics.class);
		job.setJar("export\\FlowStatistics.jar");
		// Map
		job.setMapperClass(FlowMapper.class);// 第一种
//		job.setMapperClass(FlowWritableMapper.class);
		// 这里因为同Reducer输出类型一致，可不写
//		job.setMapOutputKeyClass(Text.class);
//		job.setMapOutputValueClass(FlowWritable.class);
		// Reduce
		job.setReducerClass(FlowReducer.class);// 第一种
//		job.setReducerClass(FlowWritableReducer.class);
		// 输出k-v类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);// 第一种
//		job.setOutputValueClass(FlowWritable.class);
		
		// 设置job输入出路径
		String dataDir = "/workspace/flowStatistics/data";
		String outputDir = "/workspace/flowStatistics/output";
		Path inPath = new Path(hdfs+dataDir);
		Path outPath = new Path(hdfs+outputDir);
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outPath);
		FileSystem fs = FileSystem.get(conf);
		if(fs.exists(outPath)) {
			fs.delete(outPath, true);
		}
		
		// 运行作业
		System.out.println("Job: " + jobName + " is running...");
		if(job.waitForCompletion(true)) {
			System.out.println("success!");
			System.exit(0);
		} else {
			System.out.println("failed!");
			System.exit(1);
		}
	}
}


<|start_filename|>src/main/java/ssdut/training/mapreduce/medianstddev/MedianStdDevReducer.java<|end_filename|>
package ssdut.training.mapreduce.medianstddev;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;

public class MedianStdDevReducer extends Reducer<IntWritable, IntWritable, IntWritable, MedianStdDevTuple> {
	private MedianStdDevTuple result = new MedianStdDevTuple();			//记录评论长度中位数和标准差
	private ArrayList<Float> commentLengths = new ArrayList<Float>();	//用列表记录每条评论的长度

	@Override
	public void reduce(IntWritable key, Iterable<IntWritable> values, Context context)
			throws IOException, InterruptedException {
		float sum = 0;		//评论长度总和
		float count = 0;	//评论数
		commentLengths.clear();	//清空评论数列表
		result.setStddev(0);	//标准差默认值设为0		
		for (IntWritable val : values) {
			commentLengths.add((float) val.get());	//将评论长度保存到列表
			sum += val.get();	//计算评论长度总和
			count++;			//评论总数
		}
		
		//计算中位数：集合数量如为偶数，取中间两位的均值；如为奇数，则直接取中值
		Collections.sort(commentLengths);	//对集合中评论字数排序
		if (count % 2 == 0) {//偶
			result.setMedian((commentLengths.get((int) count / 2 - 1) + commentLengths.get((int) count / 2)) / 2.0f);
		} else {//奇
			result.setMedian(commentLengths.get((int) count / 2));
		}
		
		//计算标准差
		float mean = sum / count;	//计算评论的平均字数
		float sumOfSquares = 0.0f;	//平方和
		for (Float f : commentLengths) {
			sumOfSquares += (f - mean) * (f - mean);
		}
		result.setStddev((float) Math.sqrt(sumOfSquares / (count - 1)));	//计算标准差
		context.write(key, result);
	}
}

<|start_filename|>src/main/java/ssdut/training/mapreduce/topten/TopTenJob.java<|end_filename|>
package ssdut.training.mapreduce.topten;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
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 TopTenJob {
	public static void main(String[] args) throws Exception {
		String namenode_ip = "192.168.17.10";
		String hdfs = "hdfs://" + namenode_ip + ":9000";			
		Configuration conf = new Configuration();		
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true");
		
		String jobName = "TopTenJob";
		Job job = Job.getInstance(conf, jobName);
		job.setJarByClass(TopTenJob.class);
		job.setJar("export\\TopTen.jar");
		job.setMapperClass(TopTenMapper.class);
		job.setMapOutputKeyClass(NullWritable.class);
		job.setMapOutputValueClass(Text.class);
		job.setReducerClass(TopTenReducer.class);
		job.setOutputKeyClass(NullWritable.class);
		job.setOutputValueClass(Text.class);
		job.setNumReduceTasks(1);		//计算最终TopN，只能运行一个Reduce任务

		String dataDir = "/expr/topten/data";	
		String outputDir = "/expr/topten/output";
		Path inPath = new Path(hdfs + dataDir);
		Path outPath = new Path(hdfs + outputDir);
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outPath);		
		FileSystem fs = FileSystem.get(conf);
		if(fs.exists(outPath)) {
			fs.delete(outPath, true);
		}
		
		System.out.println( "Job: " + jobName + " is running...");
		if(job.waitForCompletion(true)) {
			System.out.println("success!");
			System.exit(0);
		} else {
			System.out.println("failed!");
			System.exit(1);
		}
	}
	
}

<|start_filename|>src/main/java/mutualFriend/JobRun.java<|end_filename|>
package mutualFriend;

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.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 JobRun {
	
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		String namenode_ip = "192.168.17.10";
		String hdfs = "hdfs://" + namenode_ip + ":9000";			
		Configuration conf = new Configuration();		
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true");
		
		// job1配置信息
		Job job1 = Job.getInstance(conf,"Decompose");
		job1.setJarByClass(JobRun.class);
		job1.setJar("export\\mutualFriend.jar");
		job1.setMapperClass(DecomposeFriendsMapper.class);
		job1.setReducerClass(DecomposeFriendsReducer.class);
		job1.setOutputKeyClass(Text.class);
		job1.setOutputValueClass(Text.class);
		
		Path input = new Path(hdfs+"/workspace/mutualFriends/data");
		Path output1 = new Path(hdfs+"/workspace/mutualFriends/output_Dec");
		FileInputFormat.addInputPath(job1, input);
		FileOutputFormat.setOutputPath(job1, output1);
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(output1)) {
			fs.delete(output1, true);
		}
		
		// job1如果运行成功则进入job2
		if(job1.waitForCompletion(true)) {//job2完全依赖job1的结果，所以job1成功执行就开启job2
			// job2配置信息
			Job job2 = Job.getInstance(conf, "Merge");
			job2.setJarByClass(JobRun.class);
			job2.setJar("export\\mutualFriend.jar");
			job2.setMapperClass(MergeFriendsMapper.class);
			job2.setReducerClass(MergeFriendsReducer.class);
			job2.setOutputKeyClass(Text.class);
			job2.setOutputValueClass(Text.class);
			
			Path output2 = new Path(hdfs+"/workspace/mutualFriends/output_Meg");
			FileInputFormat.addInputPath(job2, output1);// 输入是job1的输出
			FileOutputFormat.setOutputPath(job2, output2);
			if (fs.exists(output2)) {
				fs.delete(output2, true);
			}
			if(job2.waitForCompletion(true)) {
				System.out.println("sucessed");
			}else {
				System.out.println("failed");
			}
		}
	}
}


<|start_filename|>src/main/java/ssdut/training/mapreduce/datecount/DateSort3.java<|end_filename|>
package ssdut.training.mapreduce.datecount;

import java.io.DataInput;
import java.io.DataOutput;
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.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
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 DateSort3 {
	
	public static class MyKey implements WritableComparable<MyKey> {
		private String date;
		private int num;
		
		public String getDate() {
			return date;
		}

		public void setDate(String date) {
			this.date = date;
		}

		public int getNum() {
			return num;
		}

		public void setNum(int num) {
			this.num = num;
		}

		public MyKey() {			
		}
		
		public MyKey(String date, int num) {
			this.date = date;
			this.num = num;
		}

		@Override
		public void write(DataOutput out) throws IOException {
			out.writeUTF(date);
			out.writeInt(num);
		}

		@Override
		public void readFields(DataInput in) throws IOException {
			date = in.readUTF();
			num = in.readInt();
		}

		@Override
		public int compareTo(MyKey o) {
			//按date升序，num降序
			if (!date.equals(o.date)) //相等的话，返回true，取反为false
				return date.compareTo(o.date);
			else
				return o.num-num;
		}
	}
	
	public static class DateSort3Mapper extends Mapper<Object, Text, MyKey, NullWritable> {		
		public void map(Object key, Text value, Context context ) 
				throws IOException, InterruptedException {
			String[] strs = value.toString().split(" ");
			MyKey myKey = new MyKey(strs[0], Integer.parseInt(strs[1]));
			context.write(myKey, NullWritable.get());	//将自定义的myKey作为Map KEY输出
	    }
	}
  
	public static class DateSort3Reducer extends Reducer<MyKey,NullWritable,Text,IntWritable> {
		public void reduce(MyKey key, Iterable<NullWritable> values, Context context) 
				throws IOException, InterruptedException {			
			context.write(new Text(key.date), new IntWritable(key.num));
	    }
	}

	public static void main(String[] args) throws Exception {		
		//1.设置HDFS配置信息
		String namenode_ip = "192.168.17.10";
		String hdfs = "hdfs://" + namenode_ip + ":9000";			
		Configuration conf = new Configuration();
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true");

		//2.设置MapReduce作业配置信息
		String jobName = "DateSort3";					//定义作业名称
		Job job = Job.getInstance(conf, jobName);
		job.setJarByClass(DateSort3.class);				//指定运行时作业类
		job.setJar("export\\DateSort3.jar");			//指定本地jar包
		job.setMapperClass(DateSort3Mapper.class);		//指定Mapper类
		job.setMapOutputKeyClass(MyKey.class);			//设置Mapper输出Key类型
		job.setMapOutputValueClass(NullWritable.class);	//设置Mapper输出Value类型
		job.setReducerClass(DateSort3Reducer.class);	//指定Reducer类
		job.setOutputKeyClass(Text.class);				//设置Reduce输出Key类型
		job.setOutputValueClass(IntWritable.class); 	//设置Reduce输出Value类型
		
		//3.设置作业输入和输出路径
		String dataDir = "/expr/datecount/data";			//实验数据目录	
		String outputDir = "/expr/datecount/output_sort3";	//实验输出目录
		Path inPath = new Path(hdfs + dataDir);
		Path outPath = new Path(hdfs + outputDir);
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outPath);
		FileSystem fs = FileSystem.get(conf);
		if(fs.exists(outPath)) {
			fs.delete(outPath, true);
		}
		
		//4.运行作业
		System.out.println("Job: " + jobName + " is running...");
		if(job.waitForCompletion(true)) {
			System.out.println("success!");
			System.exit(0);
		} else {
			System.out.println("failed!");
			System.exit(1);
		}
	}

}

<|start_filename|>src/main/java/mergeMultipleFiles/MergeJob.java<|end_filename|>
package mergeMultipleFiles;

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.BytesWritable;
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;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;

import InputOutputFormatTest.MultiInOutput;

public class MergeJob {

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		// 1.设置HDFS配置信息
		String namenode_ip = "192.168.17.10";
		String hdfs = "hdfs://" + namenode_ip + ":9000";
		Configuration conf = new Configuration();
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true");

		// 2.设置MapReduce作业配置信息
		String jobName = "MergeMultipleFiles"; // 作业名称
		Job job = Job.getInstance(conf, jobName);
		job.setJarByClass(MultiInOutput.class); // 指定运行时作业类
		job.setJar("export\\MergeMultipleFiles.jar"); // 指定本地jar包
		job.setMapOutputKeyClass(Text.class); // 设置Mapper输出Key类型
		job.setMapOutputValueClass(BytesWritable.class); // 设置Mapper输出Value类型
		job.setMapperClass(MergeMapper.class);
		// 输入数据格式
		job.setInputFormatClass(MyInputFormat.class);
		// 以文件格式输出，使用序列化文件输出类
		job.setOutputFormatClass(SequenceFileOutputFormat.class);

		// 设置作业输出路径
		String inputDir = "/workspace/mergeFiles/data";
		String outputDir = "/workspace/mergeFiles/output"; // 输出目录
		Path outPath = new Path(hdfs + outputDir);
		Path inputPath = new Path(hdfs+inputDir);
		FileInputFormat.setInputPaths(job, inputPath);
		FileOutputFormat.setOutputPath(job, outPath);
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(outPath)) {
			fs.delete(outPath, true);
		}

		// 运行作业
		System.out.println("Job: " + jobName + " is running...");
		if (job.waitForCompletion(true)) {
			System.out.println("success!");
			System.exit(0);
		} else {
			System.out.println("failed!");
			System.exit(1);
		}
	}

}


<|start_filename|>src/main/java/shuffleTest/TempSort.java<|end_filename|>
package shuffleTest;

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.IntWritable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class TempSort {
	/*
	 * 按年输出（分区），每个文件包含每月的最高温度
	 */
	public static class TempSortMapper extends Mapper<Object, Text, Text, IntWritable> {
		IntWritable temp = new IntWritable();
		public void map(Object key,Text value,Context context) throws IOException, InterruptedException {
			String[] strings =value.toString().split(" ");
			String date = strings[0].substring(0, 7);
			temp.set(Integer.parseInt(strings[2].substring(0, strings[2].length()-1)));
			context.write(new Text(date), temp);
		}
	}
	
	public static class TempSortReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
		public void reduce(Text key,Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {
//			气温降序排序，区第一个
//			IntWritable temp = values.iterator().next();
//			System.out.println("气温："+temp);
//			context.write(key, temp);
			
			int  maxTemp = Integer.MIN_VALUE;
			for(IntWritable value:values) {
				System.out.println("年："+key+", 气温："+value);
				if (value.get()>maxTemp) {
					maxTemp = value.get();
				}
			}
			System.out.println("Date:"+key+", MaxTemp:"+maxTemp);
			context.write(key, new IntWritable(maxTemp));
		}
	}
	
	public static class YearPartitioner extends Partitioner<Text, IntWritable> {
		@Override
		public int getPartition(Text key, IntWritable value, int numPartitions) {
			//根据年份对数据进行分区，返回不同分区号
			if (key.toString().startsWith("1949"))
				return 0 % numPartitions;
			else if (key.toString().startsWith("1950"))
				return 1 % numPartitions;
			else
				return 2 % numPartitions;		
		}
	}
	
//	public static class MySort extends WritableComparator {
//		 public MySort() {
//			 super(IntWritable.class,true);
//		 }
//		 
//		 @SuppressWarnings({"rawtypes","unchecked"})
//		 public int compare(WritableComparable a,WritableComparable b) {
//			 return b.compareTo(a);
//		 }
//	}
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		// TODO Auto-generated method stub
		String hdfs = "hdfs://192.168.17.10:9000";
		Configuration conf = new Configuration();
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true");
		// 设置作业配置信息
		String jobName = "TempSort";
		Job job = Job.getInstance(conf, jobName);
		job.setJarByClass(TempSort.class);
		job.setJar("export\\TempSort.jar");
		// Map
		job.setMapperClass(TempSortMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		// Reduce
		job.setReducerClass(TempSortReducer.class);
		// 全局
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		// Sort
//		job.setSortComparatorClass(MySort.class);
		// Partition
		job.setPartitionerClass(YearPartitioner.class);
		job.setNumReduceTasks(3);
		//3.设置作业输入和输出路径
		String dataDir = "/expr/test/data";	//实验数据目录	
		String outputDir = "/expr/test/output";				//实验输出目录
		Path inPath = new Path(hdfs + dataDir);
		Path outPath = new Path(hdfs + outputDir);
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outPath);
		FileSystem fs = FileSystem.get(conf);
		if(fs.exists(outPath)) {
			fs.delete(outPath, true);
		}
		
		//4.运行作业
		System.out.println("Job: " + jobName + " is running...");
		if(job.waitForCompletion(true)) {
			System.out.println("success!");
			System.exit(0);
		} else {
			System.out.println("failed!");
			System.exit(1);
		}
	}

}


<|start_filename|>src/main/java/mapreduceProgram/GroupMax.java<|end_filename|>
package mapreduceProgram;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import javax.print.attribute.standard.JobName;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.RawComparator;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import mapreduceProgram.FlowSort.MySortKey;
import mapreduceProgram.FlowSort.SortMapper;
import mapreduceProgram.FlowSort.SortReducer;
import sun.tools.tree.SuperExpression;

public class GroupMax {

	public static class Pair implements WritableComparable<Pair> {
		private String order_id;
		private DoubleWritable amount;

		public Pair() {
			// TODO Auto-generated constructor stub
		}

		public Pair(String id, DoubleWritable amount) {
			this.order_id = id;
			this.amount = amount;
		}

		public String getOrder_id() {
			return order_id;
		}

		public void setOrder_id(String order_id) {
			this.order_id = order_id;
		}

		public DoubleWritable getAmount() {
			return amount;
		}

		public void setAmount(DoubleWritable amount) {
			this.amount = amount;
		}

		@Override
		public void write(DataOutput out) throws IOException {
			// TODO Auto-generated method stub
			out.writeUTF(order_id);
			out.writeDouble(amount.get());
		}

		@Override
		public void readFields(DataInput in) throws IOException {
			// TODO Auto-generated method stub
			order_id = in.readUTF();
			amount = new DoubleWritable(in.readDouble());
		}

		@Override
		public int compareTo(Pair o) {
			if (order_id.equals(o.order_id)) {// 同一order_id，按照amount降序排序
				return o.amount.compareTo(amount);
			} else {
				return order_id.compareTo(o.order_id);
			}
		}

	}

	public static class MyMapper extends Mapper<Object, Text, Pair, NullWritable> {
		Pair pair = new Pair();

		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
			String[] strs = value.toString().split(" ");
			pair.setOrder_id(strs[0]);
			pair.setAmount(new DoubleWritable(Double.parseDouble(strs[2])));
			context.write(pair, NullWritable.get());
			System.out.println(pair.getOrder_id()+","+pair.getAmount());
		}
	}

	public static class MyReducer extends Reducer<Pair, NullWritable, Text, DoubleWritable> {
		public void reduce(Pair key, Iterable<NullWritable> values, Context context)
				throws IOException, InterruptedException {
			context.write(new Text(key.getOrder_id()), key.getAmount());
			System.out.println(key.order_id+": "+key.amount.get());
			// 下面这个可以看下分组结果
//			for (NullWritable value : values) {
//				context.write(new Text(key.getOrder_id()), key.getAmount());
//				System.out.println(key.order_id+": "+key.amount.get());
//			}
		}
	}
	// 是分组不是分区，分组是组内定义一些规则由reduce去处理，分区是由多个Reduce处理，写到不同文件中
	// 自定义分组类
	public static class GroupComparator extends WritableComparator {
		public GroupComparator() {
			// TODO Auto-generated constructor stub
			super(Pair.class, true);
		}
		// Mapper端会对Pair排序，之后分组的规则是对Pair中的order_id比较
		@Override
		public int compare(WritableComparable a, WritableComparable b) {
			// TODO Auto-generated method stub
			Pair oa = (Pair) a;
			Pair ob = (Pair) b;
			return oa.getOrder_id().compareTo(ob.getOrder_id());
		}

	}

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		// 设置HDFS配置信息
		String namenode_ip = "192.168.17.10";
		String hdfs = "hdfs://" + namenode_ip + ":9000";
		Configuration conf = new Configuration();
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true");

		// 设置job配置信息
		String JobName = "GroupMax";
		Job job = Job.getInstance(conf, JobName);
		job.setJarByClass(GroupMax.class);
		job.setJar("export\\GroupMax.jar");
		// Mapper
		job.setMapperClass(MyMapper.class);
		job.setMapOutputKeyClass(Pair.class);
		job.setMapOutputValueClass(NullWritable.class);
		// Reducer
		job.setReducerClass(MyReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputKeyClass(DoubleWritable.class);
		// GroupComparator自定义分组类
		job.setGroupingComparatorClass(GroupComparator.class);
		// 作业输入输出路径
		String dataDir = "/workspace/data/orderDetail.txt"; // 实验数据
		String outputDir = "/workspace/groupMax/output"; // 实验输出目录
		Path inPath = new Path(hdfs + dataDir);
		Path outPath = new Path(hdfs + outputDir);
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outPath);
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(outPath)) {
			fs.delete(outPath, true);
		}
		// 运行作业
		System.out.println("Job: "+JobName+" is running...");
		if (job.waitForCompletion(true)) {
			System.out.println("success!");
			System.exit(0);
		} else {
			System.out.println("failed!");
			System.exit(1);
		}
	}

}


<|start_filename|>src/main/java/ssdut/training/mapreduce/itemcf/Step2.java<|end_filename|>
package ssdut.training.mapreduce.itemcf;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
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;

//计算用户评分矩阵
public class Step2 {
	public static boolean run(Configuration config, Map<String, String> paths) 
			throws IOException, ClassNotFoundException, InterruptedException {
		String jobName = "step2";
		Job job = Job.getInstance(config, jobName);
		job.setJarByClass(Step2.class);
		job.setJar("export\\ItemCF.jar");
		job.setMapperClass(Step2_Mapper.class);
		job.setReducerClass(Step2_Reducer.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);

		Path inPath = new Path(paths.get("Step2Input"));
		Path outpath = new Path(paths.get("Step2Output"));
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outpath);		
		FileSystem fs = FileSystem.get(config);
		if (fs.exists(outpath)) {
			fs.delete(outpath, true);
		}
		
		return job.waitForCompletion(true);	
	}

	static class Step2_Mapper extends Mapper<LongWritable, Text, Text, Text> {
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String[] strs = value.toString().split(",");
			String item = strs[0];		//商品id
			String user = strs[1];		//用户id
			String action = strs[2];	//用户行为
			Integer rv = StartRun.R.get(action);	//获取行为评分
			Text v = new Text(item + ":" + rv.intValue());	//value格式: "i1:1"
			Text k = new Text(user);
			context.write(k, v);	//map输出格式: "u2723  i1:1"
		}
	}

	static class Step2_Reducer extends Reducer<Text, Text, Text, Text> {
		protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
			Map<String, Integer> m = new HashMap<String, Integer>();	//用于存放每种商品的行为评分之和
			for (Text value : values) {
				String[] strs = value.toString().split(":");
				String item = strs[0];						//商品id
				Integer score = Integer.parseInt(strs[1]);	//行为评分
				score += ((Integer) (m.get(item) == null ? 0 : m.get(item))).intValue();	//计算用户对每件商品的行为评分和（如果Map集合中已有该商品评分，则累加）
				m.put(item, score);		//向HashMap中存入商品及评分之和
			}
			
			StringBuffer sb = new StringBuffer();
			for (Entry<String, Integer> entry : m.entrySet()) {
				sb.append(entry.getKey() + ":" + entry.getValue().intValue() + ",");	//将商品和评分串联，格式：  i1:1,i2:1,...I:N,
			}
			context.write(key, new Text(sb.toString().substring(0, sb.toString().length() - 1)));	//去掉最后的逗号
		}
	}
}


<|start_filename|>src/main/java/mutualFriend/DecomposeFriendsReducer.java<|end_filename|>
package mutualFriend;

import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

import org.apache.hadoop.io.Text;

public class DecomposeFriendsReducer extends Reducer<Text, Text, Text, Text>{

	@Override
	protected void reduce(Text key, Iterable<Text> values, Context context)
			throws IOException, InterruptedException {
		String friendList = "";
		for (Text value : values) {
			friendList += value.toString()+",";
		}
		// 输出个人所有好友，A	I,K,C,B,G,F,H,O,D
		context.write(key, new Text(friendList.substring(0,friendList.length()-1)));
	}

}


<|start_filename|>src/main/java/ssdut/training/mapreduce/itemcf/Step4.java<|end_filename|>
package ssdut.training.mapreduce.itemcf;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.regex.Pattern;
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.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

//同显矩阵*评分矩阵，计算评分单项
public class Step4 {
	public static boolean run(Configuration config, Map<String, String> paths) 
			throws IOException, ClassNotFoundException, InterruptedException {
		String jobName = "step4";
		Job job = Job.getInstance(config, jobName);
		job.setJarByClass(Step4.class);
		job.setJar("export\\ItemCF.jar");
		job.setMapperClass(Step4_Mapper.class);
		job.setReducerClass(Step4_Reducer.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);

		Path[] inPaths = new Path[] { 
				new Path(paths.get("Step4Input1")),
				new Path(paths.get("Step4Input2")) };		
		Path outpath = new Path(paths.get("Step4Output"));
		FileInputFormat.setInputPaths(job, inPaths);
		FileOutputFormat.setOutputPath(job, outpath);		
		FileSystem fs = FileSystem.get(config);
		if (fs.exists(outpath)) {
			fs.delete(outpath, true);
		}
		
		return job.waitForCompletion(true);
	}

	static class Step4_Mapper extends Mapper<LongWritable, Text, Text, Text> {
		private String flag;	//保存Map输入数据来自于哪个目录（output2或ouput3），用于判断数据是同现矩阵还是评分矩阵

		protected void setup(Context context) throws IOException, InterruptedException {
			FileSplit split = (FileSplit) context.getInputSplit();	//根据上下文获取输入分片对象
			flag = split.getPath().getParent().getName();			//获取输入分片所属的目录名称
		}

		protected void map(LongWritable key, Text value, Context context) 
				throws IOException, InterruptedException {
			String[] strs = Pattern.compile("[\t,]").split(value.toString());
			if (flag.equals("output3")) {				//输入的是同现矩阵，strs格式："i100:i105 1"
				String[] items = strs[0].split(":");	
				String itemID1 = items[0];				//第一个商品id  "i100"
				String itemID2 = items[1];				//第二个商品id	 "i105"
				String num = strs[1];					//两件商品的同现次数    "1"
				
				Text k = new Text(itemID1);
				Text v = new Text("A:" + itemID2 + "," + num);	//格式："A:i105,1"
				context.write(k, v);							//格式："i100	A:i105,1"
				
			} else if (flag.equals("output2")) {	//输入的是评分矩阵，strs格式："u14 i100:1 i25:1"
				String userID = strs[0];
				for (int i = 1; i < strs.length; i++) {
					String[] vector = strs[i].split(":");	//i100:1
					String itemID = vector[0];
					String score = vector[1];
					Text k = new Text(itemID);				
					Text v = new Text("B:" + userID + "," + score);	//格式："B:u14,1"
					context.write(k, v);							//格式："i100 B:u14,1" 和 "i25 B:u14,1"
				}
			}
		}
	}

	static class Step4_Reducer extends Reducer<Text, Text, Text, Text> {
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			Map<String, Integer> mapA = new HashMap<String, Integer>();
			Map<String, Integer> mapB = new HashMap<String, Integer>();
			//reduce输入格式："i100  A:i105,1  A:i107,2  B:u14,1  B:u22,3"
			for (Text val : values) {	//将AB格式的输入分别放入HashMap中
				String str = val.toString();
				if (str.startsWith("A:")) {			//str格式："A:i105,1"
					String[] kv = Pattern.compile("[\t,]").split(str.substring(2));
					mapA.put(kv[0], Integer.parseInt(kv[1]));
				} else if (str.startsWith("B:")) {	//str格式："B:u14,1"
					String[] kv = Pattern.compile("[\t,]").split(str.substring(2));
					mapB.put(kv[0], Integer.parseInt(kv[1]));
				}
			}
			double result = 0;
			Iterator<String> itera = mapA.keySet().iterator();		//根据mapA中key键(itemID)生成迭代器对象
			while (itera.hasNext()) {
				String mapka = itera.next();							//获得itemID
				int num = mapA.get(mapka).intValue();				//根据itemID从mapA获取同现次数
				
				Iterator<String> iterb = mapB.keySet().iterator();	//根据mapB中key键生成迭代器对象
				while (iterb.hasNext()) {
					String mapkb = iterb.next();					//userID
					int score = mapB.get(mapkb).intValue();			//根据userID从mapB中获取用户行为评分
					
					result = num * score;							//矩阵相乘，计算评分
					context.write(new Text(mapkb), new Text(mapka + "," + result));	//输出 key："userID" value:"itemID,result"
				}
			}
		}
	}
}


<|start_filename|>src/main/java/gradesAverage/GradesAverage.java<|end_filename|>
package gradesAverage;

import java.io.IOException;
import java.util.StringTokenizer;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.IntWritable;
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 mapReduceTest.wordCount.WordCount;
import mapReduceTest.wordCount.WordCount.IntSumReducer;
import mapReduceTest.wordCount.WordCount.TokenizerMapper;

public class GradesAverage {

	public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
		private Text student = new Text();
		private IntWritable grade = new IntWritable();

		/* (non-Javadoc)
		 * @see org.apache.hadoop.mapreduce.Mapper#map(KEYIN, VALUEIN, org.apache.hadoop.mapreduce.Mapper.Context)
		 */
		/* (non-Javadoc)
		 * @see org.apache.hadoop.mapreduce.Mapper#map(KEYIN, VALUEIN, org.apache.hadoop.mapreduce.Mapper.Context)
		 */
		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
//			StringTokenizer iTokenizer = new StringTokenizer(value.toString(),"\n");
			System.out.println("key is："+key+",value is: "+value.toString());
//			while (iTokenizer.hasMoreTokens()) {
//				
//			}
			String[] list_strs = value.toString().split(" ");
			// 因为每行只有一个学号和对应成绩，不需要考虑切分多个词
			student.set(list_strs[0]);
			grade.set(Integer.parseInt(list_strs[1]));
			context.write(student, grade);
		}
	}
	
//	public static class gradesAverageCombiner extends Reducer<Text, IntWritable, Text, Text> {
//		private Text gradesSum = new Text();
//
//		public void reduce(Text key, Iterable<IntWritable> values, Context context)
//				throws IOException, InterruptedException {
//			int sum = 0;
//			int grades = 0;
//			for (IntWritable val : values) {
//				sum += 1;
//				grades += val.get();
//			}
//			System.out.println("Combiner---student is:"+key.toString()+",grades is:"+grades+",sum is:"+sum);
//			gradesSum.set(grades+","+sum);
//			System.out.println(gradesSum);
//			context.write(key, gradesSum);
//		}
//	}
	public static class gradesAverageReducer extends Reducer<Text, IntWritable, Text, FloatWritable> {
		private FloatWritable gradesSum = new FloatWritable();

		public void reduce(Text key, Iterable<IntWritable> values, Context context)
				throws IOException, InterruptedException {
			int sum = 0;
			int grades = 0;
			for (IntWritable val : values) {
				sum += 1;   
				grades += val.get();
			}
			System.out.println("Reduce----student is:"+key.toString()+",grades is:"+grades+",sum is:"+sum);
			gradesSum.set((float)grades/sum);
			context.write(key, gradesSum);
		}
	}

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		// 1.设置HDFS配置信息
		String namenode_ip = "192.168.17.10";
		String hdfs = "hdfs://" + namenode_ip + ":9000";
		Configuration conf = new Configuration(); // Hadoop配置类
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true"); // 集群交叉提交
		/*
		 * conf.set("hadoop.job.user", "hadoop"); conf.set("mapreduce.framework.name",
		 * "yarn"); conf.set("mapreduce.jobtracker.address", namenode_ip + ":9001");
		 * conf.set("yarn.resourcemanager.hostname", namenode_ip);
		 * conf.set("yarn.resourcemanager.resource-tracker.address", namenode_ip +
		 * ":8031"); conf.set("yarn.resourcemtanager.address", namenode_ip + ":8032");
		 * conf.set("yarn.resourcemanager.admin.address", namenode_ip + ":8033");
		 * conf.set("yarn.resourcemanager.scheduler.address", namenode_ip + ":8034");
		 * conf.set("mapreduce.jobhistory.address", namenode_ip + ":10020");
		 */

		// 2.设置MapReduce作业配置信息
		String jobName = "GradesAverage"; // 定义作业名称
		Job job = Job.getInstance(conf, jobName);
		job.setJarByClass(GradesAverage.class); // 指定作业类
		job.setJar("export\\GradesAverage.jar"); // 指定本地jar包
		job.setMapperClass(TokenizerMapper.class);
//		job.setCombinerClass(gradesAverageCombiner.class); // 指定Combiner类
		job.setReducerClass(gradesAverageReducer.class);
//		输出key-value的类型
		job.setOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		job.setOutputValueClass(FloatWritable.class);

		// 3.设置作业输入和输出路径
		String dataDir = "/expr/studentgrades/grades"; // 实验数据目录
		String outputDir = "/expr/studentgrades/output"; // 实验输出目录
		Path inPath = new Path(hdfs + dataDir);
		Path outPath = new Path(hdfs + outputDir);
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outPath);
		// 如果输出目录已存在则删除
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(outPath)) {
			fs.delete(outPath, true);
		}

		// 4.运行作业
		System.out.println("Job: " + jobName + " is running...");
		if (job.waitForCompletion(true)) {
			System.out.println("统计 success!");
			System.exit(0);
		} else {
			System.out.println("统计 failed!");
			System.exit(1);
		}
	}
}


<|start_filename|>src/main/java/ssdut/training/mapreduce/minmaxcount/MinMaxCountMapper.java<|end_filename|>
package ssdut.training.mapreduce.minmaxcount;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class MinMaxCountMapper extends Mapper<Object, Text, Text, MinMaxCountTuple> {
	private Text outUserId = new Text();	//用户ID
	private MinMaxCountTuple outTuple = new MinMaxCountTuple();	//日期最小值、日期最大值、评论数的组合
	private final SimpleDateFormat frmt = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS");

	@Override
	public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
		Map<String, String> map = MRDPUtils.transformXmlToMap(value.toString());	//分解每条评论，保存每对KV到Map对象
		String userId = map.get("UserId");			//从Map对象中获取用户ID
		String strDate = map.get("CreationDate");	//从Map对象中获取评论时间
		
		if (strDate == null || userId == null) {	//过滤掉不含统计数据的记录
			return;
		}
		try {
			Date creationDate = frmt.parse(strDate);
			// 因为还没有MinMax，只有把当前数据中日期作为MinMax
			outTuple.setMin(creationDate);
			outTuple.setMax(creationDate);
			outTuple.setCount(1);
			outUserId.set(userId);
			context.write(outUserId, outTuple);
		} catch (ParseException e) {
			System.err.println(e.getMessage());
			return;
		}
	}
}


<|start_filename|>src/main/java/ssdut/training/mapreduce/topten/TopTenMapper.java<|end_filename|>
package ssdut.training.mapreduce.topten;

import java.io.IOException;
import java.util.TreeMap;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class TopTenMapper extends Mapper<Object, Text, NullWritable, Text> {
	private TreeMap<Integer, Text> visittimesMap = new TreeMap<Integer, Text>();	//TreeMap是有序KV集合

	@Override
	public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
		if (value == null) {
			return;
		}
		String[] strs = value.toString().split(" ");
		String tId = strs[0];
		String tVisittimes = strs[1];
		if (tId == null || tVisittimes == null) {
			return;
		}
		visittimesMap.put(Integer.parseInt(tVisittimes), new Text(value));	//将访问次数（KEY）和行记录（VALUE）放入TreeMap中自动排序
		if (visittimesMap.size() > 10) {	//如果TreeMap中元素超过N个，将第一个（KEY最小的）元素删除
			visittimesMap.remove(visittimesMap.firstKey());
		}
	}

	@Override
	protected void cleanup(Context context) throws IOException, InterruptedException {
		for (Text t : visittimesMap.values()) {
			context.write(NullWritable.get(), t);	//在clean()中完成Map输出
		}
	}
}


<|start_filename|>src/main/java/mapReduceTest/wordCount/WordCount.java<|end_filename|>
package mapReduceTest.wordCount;

import java.io.IOException;
import java.util.StringTokenizer;
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.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 WordCount {

	public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {		
		private final static IntWritable one = new IntWritable(1);
		private Text word = new Text();

		public void map(Object key, Text value, Context context ) throws IOException, InterruptedException {
	    	StringTokenizer itr = new StringTokenizer(value.toString());
	    	while (itr.hasMoreTokens()) {
	    		word.set(itr.nextToken());
	    		context.write(word, one);
	    	}
	    }
	}
  
	public static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> {
		private IntWritable result = new IntWritable();
		
		public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
			int sum = 0;
			for (IntWritable val : values) {
				sum += val.get();
			}
			result.set(sum);
			context.write(key, result);
	    }
	}

	public static void main(String[] args) throws Exception {		
		//1.设置HDFS配置信息
		String namenode_ip = "192.168.17.10";
		String hdfs = "hdfs://" + namenode_ip + ":9000";			
		Configuration conf = new Configuration();	//Hadoop配置类
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true");	//集群交叉提交
/*		conf.set("hadoop.job.user", "hadoop");
		conf.set("mapreduce.framework.name", "yarn");
		conf.set("mapreduce.jobtracker.address", namenode_ip + ":9001");
		conf.set("yarn.resourcemanager.hostname", namenode_ip);	
		conf.set("yarn.resourcemanager.resource-tracker.address", namenode_ip + ":8031");
		conf.set("yarn.resourcemtanager.address", namenode_ip + ":8032");
		conf.set("yarn.resourcemanager.admin.address", namenode_ip + ":8033");
		conf.set("yarn.resourcemanager.scheduler.address", namenode_ip + ":8034");
		conf.set("mapreduce.jobhistory.address", namenode_ip + ":10020"); */
		
		//2.设置MapReduce作业配置信息
		String jobName = "WordCount";					//定义作业名称
		Job job = Job.getInstance(conf, jobName);
		job.setJarByClass(WordCount.class);			//指定作业类
		job.setJar("export\\WordCount.jar");			//指定本地jar包
		job.setMapperClass(TokenizerMapper.class);
		job.setCombinerClass(IntSumReducer.class);		//指定Combiner类
		job.setReducerClass(IntSumReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		//3.设置作业输入和输出路径
		String dataDir = "/expr/wordcount/data";		//实验数据目录	
		String outputDir = "/expr/wordcount/output";	//实验输出目录
		Path inPath = new Path(hdfs + dataDir);
		Path outPath = new Path(hdfs + outputDir);
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outPath);
		//如果输出目录已存在则删除
		FileSystem fs = FileSystem.get(conf);
		if(fs.exists(outPath)) {
			fs.delete(outPath, true);
		}
		
		//4.运行作业
		System.out.println("Job: " + jobName + " is running...");
		if(job.waitForCompletion(true)) {
			System.out.println("success!");
			System.exit(0);
		} else {
			System.out.println("failed!");
			System.exit(1);
		}
	}

}

<|start_filename|>src/main/java/ssdut/training/mapreduce/topten/TopTenReducer.java<|end_filename|>
package ssdut.training.mapreduce.topten;

import java.io.IOException;
import java.util.NavigableMap;
import java.util.TreeMap;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class TopTenReducer extends Reducer<NullWritable, Text, NullWritable, Text> {
	private TreeMap<Integer, Text> visittimesMap = new TreeMap<Integer, Text>();

	@Override
	public void reduce(NullWritable key, Iterable<Text> values, Context context)
			throws IOException, InterruptedException {
		for (Text val : values) {
			String[] strs = val.toString().split(" ");
			visittimesMap.put(Integer.parseInt(strs[1]), new Text(val));	//将访问次数（KEY）和行记录（VALUE）放入TreeMap中自动排序
			if (visittimesMap.size() > 10) {		//如果TreeMap中元素超过N个，将第一个（KEY最小的）元素删除
				visittimesMap.remove(visittimesMap.firstKey());
			}
		}
	}
	
	public void cleanup(Context context) throws IOException, InterruptedException {
		//将TreeMap反序处理，降序输出top10
		NavigableMap<Integer, Text> reverMap = visittimesMap.descendingMap();	//获得TreeMap反序
		for (Text t : reverMap.values()) {
			context.write(NullWritable.get(), t);
		}
	}
}


 


<|start_filename|>src/main/java/mergeMultipleFiles/MyRecordReader.java<|end_filename|>
package mergeMultipleFiles;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

public class MyRecordReader extends RecordReader<NullWritable, BytesWritable>{
	private FileSplit fileSplit;
	private Configuration conf ;
	private BytesWritable value = new BytesWritable();
	private boolean processed =false;
	@Override
	public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
		fileSplit = (FileSplit)split;
		conf = context.getConfiguration();
	}
	@Override
	public boolean nextKeyValue() throws IOException, InterruptedException {
		if (!processed) {
			byte[] contents = new byte[(int) fileSplit.getLength()];// 获取分片长度字节数组
			Path file = fileSplit.getPath();// 获取切片所在位置
			FileSystem fSystem = file.getFileSystem(conf);
			FSDataInputStream in = null;
			try {
				in = fSystem.open(file);// 打开文件
				IOUtils.readFully(in, contents, 0, contents.length);// 读取整个文件字节数据，写入contents
				value.set(contents,0,contents.length);// 将整个文件数据赋值给value
			} finally {
				IOUtils.closeStream(in);
			}
			processed = true;
			return true;
		}
		return false;
	}
	@Override
	public NullWritable getCurrentKey() throws IOException, InterruptedException {
		// 获取当前key，因为合并文件，我们应该将文件对象付给value，key赋空即可
		return NullWritable.get();
	}
	@Override
	public BytesWritable getCurrentValue() throws IOException, InterruptedException {
		return value;// value是整个文件对象的字节数据
	}
	@Override
	public float getProgress() throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		return processed ? 1.0f:0.0f;
	}
	@Override
	public void close() throws IOException {
		// TODO Auto-generated method stub
		
	}
}


<|start_filename|>src/main/java/mergeMultipleFiles/MergeMapper.java<|end_filename|>
package mergeMultipleFiles;

import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

public class MergeMapper extends Mapper<NullWritable, BytesWritable, Text, BytesWritable>{
	private Text fileNameKey;

	@Override
	protected void map(NullWritable key, BytesWritable value,
			Mapper<NullWritable, BytesWritable, Text, BytesWritable>.Context context)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		context.write(fileNameKey, value);
	}

	@Override
	protected void setup(Mapper<NullWritable, BytesWritable, Text, BytesWritable>.Context context)
			throws IOException, InterruptedException {
		InputSplit split = context.getInputSplit();
		Path path = ((FileSplit)split).getPath();//???
		fileNameKey = new Text(path.toString());
	}
	
}


<|start_filename|>src/main/java/mutualFriend/JobControlRun.java<|end_filename|>
package mutualFriend;

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.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.jobcontrol.ControlledJob;
import org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class JobControlRun {

	public static void main(String[] args) throws IOException {
		String namenode_ip = "192.168.17.10";
		String hdfs = "hdfs://" + namenode_ip + ":9000";			
		Configuration conf = new Configuration();		
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true");
		
		Job job1 = Job.getInstance(conf,"Decompose");
		job1.setJarByClass(JobControlRun.class);
		job1.setJar("export\\mutualFriend.jar");
		job1.setMapperClass(DecomposeFriendsMapper.class);
		job1.setReducerClass(DecomposeFriendsReducer.class);
		job1.setOutputKeyClass(Text.class);
		job1.setOutputValueClass(Text.class);
		
		Path input = new Path(hdfs+"/workspace/mutualFriends/data");
		Path output1 = new Path(hdfs+"/workspace/mutualFriends/output_Dec");
		FileInputFormat.addInputPath(job1, input);
		FileOutputFormat.setOutputPath(job1, output1);
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(output1)) {
			fs.delete(output1, true);
			System.out.println("我被删了");// 打印可见只被删了一次，有点怪
		}
		// ControlledJob作业控制容器
		ControlledJob ctrJob1=new ControlledJob(conf);
		ctrJob1.setJob(job1);// job1加入控制容器
		
		Job job2 = Job.getInstance(conf, "Merge");
		job2.setJarByClass(JobControlRun.class);
		job2.setJar("export\\mutualFriend.jar");
		job2.setMapperClass(MergeFriendsMapper.class);
		job2.setReducerClass(MergeFriendsReducer.class);
		job2.setOutputKeyClass(Text.class);
		job2.setOutputValueClass(Text.class);
		
		Path input2 = new Path(hdfs+"/workspace/mutualFriends/output_Dec");
		Path output2 = new Path(hdfs+"/workspace/mutualFriends/output_Meg");
		FileInputFormat.addInputPath(job2, input2);
		FileOutputFormat.setOutputPath(job2, output2);
		if (fs.exists(output2)) {
			fs.delete(output2, true);
		}
		ControlledJob ctrJob2 = new ControlledJob(conf);
		ctrJob2.setJob(job2);// job2加入作业控制容器
		
		// 添加作业依赖，表明job2依赖job1执行
		ctrJob2.addDependingJob(ctrJob1);
		
		// 定义作业主控制容器，监控、调度job1，job2
		JobControl jobControl=new JobControl("JobControl");
		jobControl.addJob(ctrJob1);
		jobControl.addJob(ctrJob2);
		// 启动作业线程
		Thread T=new Thread(jobControl);
		T.start();
		while(true){
			if(jobControl.allFinished()){// 等待作业全部结束
				System.out.println(jobControl.getSuccessfulJobList());// 打印成功job信息
				jobControl.stop();
				break;
			}
		}
		/**
		 * 打印控制信息如下
		 * [job name:	Decompose
			job id:	JobControl0
			job state:	SUCCESS
			job mapred id:	job_local445604445_0001
			job message:	just initialized
			job has no depending job:	
			, job name:	Merge
			job id:	JobControl1
			job state:	SUCCESS
			job mapred id:	job_local1897659504_0002
			job message:	just initialized
			job has 1 dependeng jobs:
				 depending job 0:	Decompose
			]
		 */
	}

}


<|start_filename|>src/main/java/ssdut/training/mapreduce/itemcf/Step1.java<|end_filename|>
package ssdut.training.mapreduce.itemcf;

import java.io.IOException;
import java.util.Map;
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.NullWritable;
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 Step1 {
	public static boolean run(Configuration config, Map<String, String> paths)
			throws IllegalArgumentException, IOException, ClassNotFoundException, InterruptedException {
		String jobName = "step1";
		Job job = Job.getInstance(config, jobName);
		job.setJarByClass(Step1.class);
		job.setJar("export\\ItemCF.jar");
		job.setMapperClass(Step1_Mapper.class);
		job.setReducerClass(Step1_Reducer.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(NullWritable.class);

		Path inPath = new Path(paths.get("Step1Input"));
		Path outpath = new Path(paths.get("Step1Output"));
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outpath);		
		FileSystem fs = FileSystem.get(config);
		if (fs.exists(outpath)) {
			fs.delete(outpath, true);
		}
		
		return job.waitForCompletion(true);
	}

	static class Step1_Mapper extends Mapper<LongWritable, Text, Text, NullWritable> {
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			if (key.get() != 0) {	//过滤掉输入文件标题行
				context.write(value, NullWritable.get());
			}
		}
	}

	static class Step1_Reducer extends Reducer<Text, NullWritable, Text, NullWritable> {
		protected void reduce(Text key, Iterable<NullWritable> values, Context context)
				throws IOException, InterruptedException {
			context.write(key, NullWritable.get());
		}
	}
}


<|start_filename|>src/main/java/mapreduceProgram/FlowPartition.java<|end_filename|>
package mapreduceProgram;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.HashMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class FlowPartition {
	public static class FlowPartitionMapper extends Mapper<Object, Text, Text, FlowWritable> {
		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
			String[] strs = value.toString().split("\t");
			Text phone = new Text(strs[0]);
			FlowWritable flow = new FlowWritable(Integer.parseInt(strs[1]), Integer.parseInt(strs[2]));
			System.out.println("Flow is:" + flow.toString());
			context.write(phone, flow);
		}
	}

	public static class FlowPartitionReducer extends Reducer<Text, FlowWritable, Text, FlowWritable> {
		public void reduce(Text key, Iterable<FlowWritable> values, Context context)
				throws IOException, InterruptedException {
			int upFlow = 0;
			int downFlow = 0;

			for (FlowWritable value : values) {
				upFlow += value.getUpFlow();
				downFlow += value.getDownFlow();
			}
			System.out.println(key.toString() + ":" + upFlow + "," + downFlow);
			context.write(key, new FlowWritable(upFlow, downFlow));
		}
	}

	public static class FlowWritable implements Writable {
		private int upFlow;
		private int downFlow;
		private int sumFlow;

		public FlowWritable() {
		}

		public FlowWritable(int upFlow, int downFlow) {
			this.upFlow = upFlow;
			this.downFlow = downFlow;
			this.sumFlow = upFlow + downFlow;
		}

		public int getDownFlow() {
			return downFlow;
		}

		public void setDownFlow(int downFlow) {
			this.downFlow = downFlow;
		}

		public int getUpFlow() {
			return upFlow;
		}

		public void setUpFlow(int upFlow) {
			this.upFlow = upFlow;
		}

		public int getSumFlow() {
			return sumFlow;
		}

		public void setSumFlow(int sumFlow) {
			this.sumFlow = sumFlow;
		}

		@Override
		public void write(DataOutput out) throws IOException {
			// TODO Auto-generated method stub
			out.writeInt(upFlow);
			out.writeInt(downFlow);
			out.writeInt(sumFlow);
		}

		@Override
		public void readFields(DataInput in) throws IOException {
			// TODO Auto-generated method stub
			upFlow = in.readInt();
			downFlow = in.readInt();
			sumFlow = in.readInt();
		}

		@Override
		public String toString() {
			// TODO Auto-generated method stub
			return upFlow + "\t" + downFlow + "\t" + sumFlow;
		}
	}

	public static class PhoneNumberPartitioner extends Partitioner<Text, FlowWritable> {
		private static HashMap<String, Integer> numberDict = new HashMap<>();
		static {
			numberDict.put("133", 0);
			numberDict.put("135", 1);
			numberDict.put("137", 2);
			numberDict.put("138", 3);
		}

		@Override
		public int getPartition(Text key, FlowWritable value, int numPartitions) {
			String num = key.toString().substring(0, 3);
			// 借助HashMap返回不同手机段对应的分区号
			// 也可以直接通过if判断，如
			// 根据年份对数据进行分区，返回不同分区号
			// if (key.toString().startsWith("133")) return 0 % numPartitions;
			return numberDict.getOrDefault(num, 4);
		}
	}

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		// 设置hdfs配置信息
		String namenode_ip = "192.168.17.10";
		String hdfs = "hdfs://" + namenode_ip + ":9000";
		Configuration conf = new Configuration();
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true");

		// 设置作业Job配置信息
		String jobName = "FlowPartition";
		Job job = Job.getInstance(conf, jobName);
		job.setJarByClass(FlowPartition.class);
		job.setJar("export\\FlowPartition.jar");
		// Map
		 job.setMapperClass(FlowPartitionMapper.class);
		// Reduce
		job.setReducerClass(FlowPartitionReducer.class);
		// 输出k-v类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(FlowWritable.class);
		// 设置分区类,及Reducer数目
		job.setPartitionerClass(PhoneNumberPartitioner.class);
		job.setNumReduceTasks(4);

		// 设置job输入出路径
		String dataDir = "/workspace/flowStatistics/data";
		String outputDir = "/workspace/flowStatistics/output_partitions";
		Path inPath = new Path(hdfs + dataDir);
		Path outPath = new Path(hdfs + outputDir);
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outPath);
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(outPath)) {
			fs.delete(outPath, true);
		}

		// 运行作业
		System.out.println("Job: " + jobName + " is running...");
		if (job.waitForCompletion(true)) {
			System.out.println("success!");
			System.exit(0);
		} else {
			System.out.println("failed!");
			System.exit(1);
		}
	}
}


<|start_filename|>src/main/java/ssdut/training/mapreduce/inputformat/FixedLengthInput.java<|end_filename|>
package ssdut.training.mapreduce.inputformat;

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.BytesWritable;
import org.apache.hadoop.io.LongWritable;
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.input.FixedLengthInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class FixedLengthInput {
	public static class FixedLengthInputMapper extends Mapper<LongWritable, BytesWritable, LongWritable, BytesWritable> {		
		public void map(LongWritable key, BytesWritable value, Context context ) 
				throws IOException, InterruptedException {
			context.write(key, value);
	    }
	}
  
	public static class FixedLengthInputReducer extends Reducer<LongWritable,BytesWritable,LongWritable,BytesWritable> {
		public void reduce(LongWritable key, Iterable<BytesWritable> values, Context context) 
				throws IOException, InterruptedException {
			for (BytesWritable val : values) {
				context.write(key, val);
			}
	    }
	}

	public static void main(String[] args) throws Exception {		
		//1.设置HDFS配置信息
		String namenode_ip = "192.168.17.10";
		String hdfs = "hdfs://" + namenode_ip + ":9000";			
		Configuration conf = new Configuration();
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true");
		conf.setInt(FixedLengthInputFormat.FIXED_RECORD_LENGTH, 13);
		
		//2.设置MapReduce作业配置信息
		String jobName = "FixedLengthInput";					//作业名称
		Job job = Job.getInstance(conf, jobName);
		job.setJarByClass(FixedLengthInput.class);				//指定运行时作业类
		job.setJar("export\\FixedLengthInput.jar");				//指定本地jar包
		job.setMapperClass(FixedLengthInputMapper.class);		//指定Mapper类
		job.setMapOutputKeyClass(LongWritable.class);			//设置Mapper输出Key类型
		job.setMapOutputValueClass(BytesWritable.class);		//设置Mapper输出Value类型
		job.setReducerClass(FixedLengthInputReducer.class);		//指定Reducer类
		job.setOutputKeyClass(LongWritable.class);				//设置Reduce输出Key类型
		job.setOutputValueClass(BytesWritable.class); 			//设置Reduce输出Value类型
		
		job.setInputFormatClass(FixedLengthInputFormat.class);	//设置输入格式化类
		
		//3.设置作业输入和输出路径
		String dataDir = "/expr/fixedinput/data";			//实验数据目录	
		String outputDir = "/expr/fixedinput/output";		//实验输出目录
		Path inPath = new Path(hdfs + dataDir);
		Path outPath = new Path(hdfs + outputDir);
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outPath);
		FileSystem fs = FileSystem.get(conf);
		if(fs.exists(outPath)) {
			fs.delete(outPath, true);
		}
		
		//4.运行作业
		System.out.println("Job: " + jobName + " is running...");
		if(job.waitForCompletion(true)) {
			System.out.println("success!");
			System.exit(0);
		} else {
			System.out.println("failed!");
			System.exit(1);
		}
	}

}

<|start_filename|>src/main/java/mutualFriend/MergeFriendsReducer.java<|end_filename|>
package mutualFriend;

import java.io.IOException;

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

public class MergeFriendsReducer extends Reducer<Text, Text, Text, Text>{
	public void reduce(Text key,Iterable<Text> values,Context context) throws IOException, InterruptedException {
		String friends = "";
		for (Text value : values) {
			friends += value.toString()+",";
		}
		System.out.println(key.toString()+" "+friends);
		context.write(key, new Text(friends));
	}
}


<|start_filename|>src/main/java/mergeMultipleFiles/MyInputFormat.java<|end_filename|>
package mergeMultipleFiles;

import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;


public class MyInputFormat extends FileInputFormat<NullWritable, BytesWritable>{

	
	@Override
	protected boolean isSplitable(JobContext context, Path filename) {
		// TODO 因为是合并小文件，设置文件不可分割，k-v的v就是文件对象
		return false;
	}

	@Override
	public RecordReader<NullWritable, BytesWritable> createRecordReader(InputSplit split, TaskAttemptContext context)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		MyRecordReader myRecordReader = new MyRecordReader();
		myRecordReader.initialize(split, context);
		return myRecordReader;
	}

}


<|start_filename|>src/main/java/mutualFriend/MergeFriendsMapper.java<|end_filename|>
package mutualFriend;

import java.io.IOException;
import java.util.Arrays;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class MergeFriendsMapper extends Mapper<Object, Text, Text, Text>{// 别写成输入key也是Text类型，这里输入的是偏移量
	public void map(Object key,Text value,Context context) throws IOException, InterruptedException {
		Text uText = new Text(value.toString().substring(0, 1));
		String[] lists = value.toString().substring(2).split(",");
		Arrays.sort(lists);// 要排好序，不然如A-B，B-A不能归并到一起
		//对如A B,C,E遍历输出如<B-C A>
		for (int i = 0; i < lists.length; i++) {
			for(int j=i+1;j<lists.length;j++) {
				String friends = lists[i]+"-"+lists[j];
//				System.out.println(friends);
				context.write(new Text(friends), uText);
			}
		}
	}
}


<|start_filename|>src/main/java/ssdut/training/mapreduce/datecount/DateFilter.java<|end_filename|>
package ssdut.training.mapreduce.datecount;

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.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class DateFilter {

	public static class DateFilterMapper extends Mapper<Object, Text, Text, NullWritable> {		
		public void map(Object key, Text value, Context context ) throws IOException, InterruptedException {
	    	String[] strs = value.toString().split(" ");
	    	Text date = new Text(strs[0]);
			context.write(date, NullWritable.get());
	    }
	}
/*
	public static class DateFilterReducer extends Reducer<Text,IntWritable,Text,IntWritable> {
		public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
			int sum = 0;
			for (IntWritable val : values) {
				sum += val.get();
			}
			context.write(key, new IntWritable(sum));
	    }
	}
*/
	public static void main(String[] args) throws Exception {		
		//1.设置HDFS配置信息
		String namenode_ip = "192.168.17.10";
		String hdfs = "hdfs://" + namenode_ip + ":9000";			
		Configuration conf = new Configuration();
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true");
		
		//2.设置MapReduce作业配置信息
		String jobName = "DateFilter";					//定义作业名称
		Job job = Job.getInstance(conf, jobName);		
		job.setJarByClass(DateFilter.class);			//指定运行时作业类
		job.setJar("export\\DateFilter.jar");			//指定本地jar包
		job.setMapperClass(DateFilterMapper.class);
		job.setMapOutputKeyClass(Text.class);			//设置Mapper输出Key类型
		job.setMapOutputValueClass(NullWritable.class);	//设置Mapper输出Value类型
		//job.setReducerClass(DateCountReducer.class);	//不需要设置Reducer类
		//job.setOutputKeyClass(Text.class);			//设置Reduce输出键类型
		//job.setOutputValueClass(NullWritable.class);	//设置Reduce输出值类型
		
		//3.设置作业输入和输出路径
		String dataDir = "/expr/datecount/data";			//实验数据目录	
		String outputDir = "/expr/datecount/output_filter";	//实验输出目录
		Path inPath = new Path(hdfs + dataDir);
		Path outPath = new Path(hdfs + outputDir);
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outPath);
		FileSystem fs = FileSystem.get(conf);
		if(fs.exists(outPath)) {
			fs.delete(outPath, true);
		}
		
		//4.运行作业
		System.out.println("Job: " + jobName + " is running...");
		if(job.waitForCompletion(true)) {
			System.out.println("success!");
			System.exit(0);
		} else {
			System.out.println("failed!");
			System.exit(1);
		}
	}

}

<|start_filename|>src/main/java/shuffleTest/MonthAscTempDescSort.java<|end_filename|>
package shuffleTest;

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.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Mapper.Context;

public class MonthAscTempDescSort {
//	按年分区，每个文件中按月升序，按温度降序
	public static class MonthTempMapper extends Mapper<Object, Text, Text, IntWritable> {
		IntWritable temp = new IntWritable();
		public void map(Object key,Text value,Context context) throws IOException, InterruptedException {
			String[] strings =value.toString().split(" ");
			String date = strings[0].substring(0, 7);
			temp.set(Integer.parseInt(strings[2].substring(0, strings[2].length()-1)));
			context.write(new Text(date), temp);
		}
	}
	

	public static void main(String[] args) {
		// TODO Auto-generated method stub

	}

}


<|start_filename|>src/main/java/ssdut/training/mapreduce/peoplerank/People.java<|end_filename|>
package ssdut.training.mapreduce.peoplerank;

import java.io.IOException;
import java.util.Arrays;
import org.apache.commons.lang.StringUtils;

public class People {
	private double peopleRank = 1.0;		//存储 PR值，初值默认1.0
	private String[] attentionPeoples;		//关注的人
	public static final char fieldSeparator = '\t';	//多处使用分隔符\t，定义为常量

	public double getPeopleRank() {
		return peopleRank;
	}

	public People setPeopleRank(double pageRank) {
		this.peopleRank = pageRank;
		return this;
	}

	public String[] getAttentionPeoples() {
		return attentionPeoples;
	}

	public People setAttentionPeoples(String[] attentionPeoples) {
		this.attentionPeoples = attentionPeoples;
		return this;
	}

	//判断是否包含关注用户
	public boolean containsAttentionPeoples() {
		return attentionPeoples != null && attentionPeoples.length > 0;
	}

	@Override
	//People对象转成字符串
	public String toString() {
		StringBuilder sb = new StringBuilder();
		sb.append(peopleRank);
		if (attentionPeoples != null) {
			sb.append(fieldSeparator).append(StringUtils.join(attentionPeoples, fieldSeparator));
		}
		return sb.toString();	//返回String格式："PeopleRand值	u1	u2..."
	}
	
	//字符串转成People对象
	public static People fromMR(String str) throws IOException {	//参数String格式："PeopleRand值	u1	u2..."
		People people = new People();
		String[] strs = StringUtils.splitPreserveAllTokens(str, fieldSeparator);	//将字符串按分隔符分割成字符串数组
		people.setPeopleRank(Double.valueOf(strs[0]));	//处理第一个元素
		if (strs.length > 1) {// 设置关注的人，从strs下标为1的位置开始（因为传进来类似"1.0 b c d"的字符串）
			people.setAttentionPeoples(Arrays.copyOfRange(strs, 1, strs.length));	//处理其它元素
		}
		return people;	//返回People对象
	}
}


<|start_filename|>src/main/java/ssdut/training/mapreduce/peoplerank/PeopleRank2.java<|end_filename|>
package ssdut.training.mapreduce.peoplerank;

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.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.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class PeopleRank2 {
	public static enum Mycounter {
		my
	}

	static class PeopleRank2Mapper extends Mapper<Text, Text, Text, Text> {
		protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
			String pid = key.toString();	//key:"a"  value:"d c b"
			int runCount = context.getConfiguration().getInt("runCount", 1);	//从上下文配置中获取runCount的值，如果值为空则默认为1
			People people = null;
			if (runCount == 1) {	//第一次运行时，Map输入记录中没有PR值，将PR值默认设为1.0
				people = People.fromMR("1.0" + People.fieldSeparator + value.toString());//参数String格式："PeopleRank值	u1	u2..."
			} else {				//后续迭代的Map输入value中已经包含PR值，无需再指定
				people = People.fromMR(value.toString());
			}
			context.write(new Text(pid), new Text(people.toString())); 		//Map输出格式："userid pr值  userlist"
			
			if (people.containsAttentionPeoples()) {	//如果Map输入中有被关注人，则计算每个被关注人的概率并通过Map输出
				double outValue = people.getPeopleRank() / people.getAttentionPeoples().length;			//例如：a关注bcd，则bcd得到的概率都是1.0/3
				for (int i = 0; i < people.getAttentionPeoples().length; i++) {
					context.write(new Text(people.getAttentionPeoples()[i]), new Text(outValue + ""));	//Map输出格式："被关注人id 关注人投给被关注人的概率"
				}
			}
		}
	}

	static class PeopleRank2Reducer extends Reducer<Text, Text, Text, Text> {
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			double sum = 0.0;
			People sourcePeople = null;
			for (Text v : values) {
				People people = People.fromMR(v.toString());
				if (people.containsAttentionPeoples()) {		//含有关注人的是源结构
					sourcePeople = people;
				} else {
					sum += people.getPeopleRank();				//将不含临近节点的PR值累加
				}
			}
			double newPR = (1 - 0.85) / 4 + (0.85 * sum);		//计算新的PR值 = (1-d)/N + d* sum（每个被关注者PR值/每个关注人链出数）  (阻尼系数取0.85)
			double d = newPR - sourcePeople.getPeopleRank();	//计算新旧PR差值
			int j = Math.abs( (int)(d*100.0) );					//取收敛差值放大100倍后的绝对值
			context.getCounter(Mycounter.my).increment(j);		//放入计数器
			sourcePeople.setPeopleRank(newPR);					//更新PR值			
			context.write(key, new Text(sourcePeople.toString()));	//输出格式："userid  新PR值  userlist"
		}
	}
	
	//查看Reduce输入
	static class TestReducer extends Reducer<Text, Text, Text, Text> {
		protected void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			StringBuilder sb = new StringBuilder();
			for (Text v : values) {
				sb.append("["+v.toString()+"] ");
			}
			context.write(key, new Text(sb.toString()));
		}
	}
	
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		String namenode_ip = "192.168.17.10";
		String hdfs = "hdfs://" + namenode_ip + ":9000";			
		Configuration conf = new Configuration();		
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true");
		conf.set("mapreduce.framework.name", "yarn");
		conf.set("yarn.resourcemanager.hostname", namenode_ip);	
		conf.set("yarn.resourcemanager.scheduler.address", namenode_ip + ":8034");
		//conf.set("hadoop.job.user", "hadoop");		
		//conf.set("yarn.resourcemanager.resource-tracker.address", namenode_ip + ":8031");
		//conf.set("yarn.resourcemtanager.address", namenode_ip + ":8032");
		//conf.set("yarn.resourcemanager.admin.address", namenode_ip + ":8033");
		//conf.set("mapreduce.jobtracker.address", namenode_ip + ":9001");
		//conf.set("mapreduce.jobhistory.address", namenode_ip + ":10020");		
		//conf.set("mapreduce.map.memory.mb", "2048");
		//conf.set("mapreduce.task.io.sort.mb","300");
		
		String jobName = "PeopleRank";
		double e = 0.01;	//收敛差的阈值，e越小迭代次数越多
		int i = 0;
		while (true) {
			i++;
			conf.setInt("runCount", i);		//runCount记录迭代次数
			Job job = Job.getInstance(conf, jobName + i);
			job.setJarByClass(PeopleRank2.class);
			job.setJar("export\\PeopleRank2.jar");
			job.setMapperClass(PeopleRank2Mapper.class);
			job.setReducerClass(PeopleRank2Reducer.class);
			//job.setReducerClass(TestReducer.class);
			job.setMapOutputKeyClass(Text.class);
			job.setMapOutputValueClass(Text.class);
			job.setInputFormatClass(KeyValueTextInputFormat.class);		//设置Map输入为KV文本文件类型
			
			Path inPath = null;
			if (i == 1) {
				inPath = new Path(hdfs + "/expr/peoplerank/output/adjacent");
			} else {
				inPath = new Path(hdfs + "/expr/peoplerank/output/output" + (i - 1));	//每次迭代使用上一次的输出路径作为输入路径
			}
			Path outPath = new Path(hdfs + "/expr/peoplerank/output/output" + i);
			FileInputFormat.addInputPath(job, inPath);
			FileOutputFormat.setOutputPath(job, outPath);			
			FileSystem fs = FileSystem.get(conf);				
			if (fs.exists(outPath)) {
				fs.delete(outPath, true);
			}
			
			System.out.println("Job Iteration " + i + " is running...");
			boolean f = job.waitForCompletion(true);
			if (f) {
				long sum = job.getCounters().findCounter(Mycounter.my).getValue();
				double avge = sum / 400.0;	//sum值在reduce任务中放大了100倍，此处需缩小100倍
				if (avge < e) {				//当平均差值小于阈值时迭代结束
					break;
				}
			}
		}
		System.out.println("finished!");
	}

}


<|start_filename|>src/main/java/InvertedIndex/InvertedCombiner.java<|end_filename|>
package InvertedIndex;

import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class InvertedCombiner extends Reducer<Text, Text, Text, Text> {
	private Text info = new Text();

	public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
		int sum = 0;
		for (Text value : values) {
			sum += Integer.parseInt(value.toString());	//单词数求和
		}
		int splitIndex = key.toString().indexOf(":");	//获取key中冒号的下标
		//注意此处应先计算info再计算key，否则key下标会越界
		info.set(key.toString().substring(splitIndex + 1) + ":" + sum);	//将key中冒号后的内容（文件名）与单词数总和组合成Combiner输出的value
		key.set(key.toString().substring(0, splitIndex));				//将key中冒号前的内容（单词）设置为Combiner输出的key
		context.write(key, info);		//输出格式："word	filename:sum"
	}
}


<|start_filename|>src/main/java/ssdut/training/mapreduce/medianstddev/MedianStdDevMapper.java<|end_filename|>
package ssdut.training.mapreduce.medianstddev;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.Map;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class MedianStdDevMapper extends Mapper<Object, Text, IntWritable, IntWritable> {
	private IntWritable outHour= new IntWritable();
	private IntWritable outCommentLength= new IntWritable();
	private final static SimpleDateFormat frmt = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS");

	@SuppressWarnings("deprecation")
	@Override
	public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
		Map<String, String> map = MRDPUtils.transformXmlToMap(value.toString());
		String strDate = map.get("CreationDate");		//获取评论日期
		String text = map.get("Text");					//获取评论内容
		if (strDate == null || text == null) {
			return;
		}
		try {
			Date creationDate = frmt.parse(strDate);	//转换日期格式
			outHour.set(creationDate.getHours());		//从日期中获取小时值
			outCommentLength.set(text.length());		//设置评论内容的长度
			context.write(outHour, outCommentLength);	//将小时和评论长度作为Map输出
		} catch (ParseException e) {
			System.err.println(e.getMessage());
			return;
		}
	}
}

<|start_filename|>src/main/java/ssdut/training/mapreduce/itemcf/Step3.java<|end_filename|>
package ssdut.training.mapreduce.itemcf;

import java.io.IOException;
import java.util.Map;
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.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 Step3 {
	private final static Text K = new Text();
	private final static IntWritable one = new IntWritable(1);

	public static boolean run(Configuration config, Map<String, String> paths)  throws IOException, ClassNotFoundException, InterruptedException {
		String jobName = "step3";
		Job job = Job.getInstance(config, jobName);
		job.setJarByClass(Step3.class);
		job.setJar("export\\ItemCF.jar");
		job.setMapperClass(Step3_Mapper.class);
		job.setReducerClass(Step3_Reducer.class);
		job.setCombinerClass(Step3_Reducer.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);

		Path inPath = new Path(paths.get("Step3Input"));
		Path outpath = new Path(paths.get("Step3Output"));
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outpath);		
		FileSystem fs = FileSystem.get(config);
		if (fs.exists(outpath)) {
			fs.delete(outpath, true);
		}
		
		return job.waitForCompletion(true);		
	}

	static class Step3_Mapper extends Mapper<LongWritable, Text, Text, IntWritable> {
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			// u2727 i468:2,i446:3			
			String[] items = value.toString().split("\t")[1].split(",");			//每件商品和评分列表，格式：i468:2 i446:3
			for (int i = 0; i < items.length; i++) {
				String itemA = items[i].split(":")[0];		// itemA = i468 .. i446
				for (int j = 0; j < items.length; j++) {
					String itemB = items[j].split(":")[0];	// itemB = i468 .. i446
					K.set(itemA + ":" + itemB);				// i468:i468 , i468:i446, i446:i468, i446:i446
					context.write(K, one);
				}
			}
		}
	}

	static class Step3_Reducer extends Reducer<Text, IntWritable, Text, IntWritable> {
		protected void reduce(Text key, Iterable<IntWritable> values, Context context)
				throws IOException, InterruptedException {
			int sum = 0;
			for (IntWritable val : values) {
				sum += val.get();
			}
			context.write(key, new IntWritable(sum));
		}
	}
}


<|start_filename|>src/main/java/ssdut/training/mapreduce/datecount/DateSort.java<|end_filename|>
package ssdut.training.mapreduce.datecount;

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.IntWritable;
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 DateSort {	
	public static class DateSortMapper extends Mapper<Object, Text, IntWritable, Text> {	//key-value类型不同于以往的	
		IntWritable num = new IntWritable();
		public void map(Object key, Text value, Context context ) throws IOException, InterruptedException {
	    	String[] strs = value.toString().split("\t");	//从DateCount运行结果读取数据，默认是用Tab分割输入
	    	String date = strs[0];					//获取日期
	    	num.set(Integer.parseInt(strs[1]));		//获取次数
	    	context.write(num, new Text(date));		//以次数作为key，日期作为value输出;利用shuffle自动对key升序排序的特性
	    }
	}
  
	public static class DateSortReducer extends Reducer<IntWritable,Text,Text,IntWritable> {
		public void reduce(IntWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
			for (Text val : values) {
				context.write(val, key);//Map阶段将日期和次数反过来以实现排序，Reduce这里再次翻转key-value
			}
	    }
	}

	public static void main(String[] args) throws Exception {		
		//1.设置HDFS配置信息
		String namenode_ip = "192.168.17.10";
		String hdfs = "hdfs://" + namenode_ip + ":9000";			
		Configuration conf = new Configuration();
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true");

		//2.设置MapReduce作业配置信息
		String jobName = "DateSort";					//定义作业名称
		Job job = Job.getInstance(conf, jobName);
		job.setJarByClass(DateSort.class);				//指定作业类
		job.setJar("export\\DateSort.jar");				//指定本地jar包
		job.setMapperClass(DateSortMapper.class);		//指定Mapper类
		job.setMapOutputKeyClass(IntWritable.class);	//设置Mapper输出Key类型
		job.setMapOutputValueClass(Text.class);			//设置Mapper输出Value类型
		job.setReducerClass(DateSortReducer.class);		//指定Reducer类
		job.setOutputKeyClass(Text.class);				//设置Reduce输出Key类型
		job.setOutputValueClass(IntWritable.class);		//设置Reduce输出Value类型
				
		//3.设置作业输入和输出路径
		String dataDir = "/expr/datecount/output/part-r-00000";	//实验数据目录	
		String outputDir = "/expr/datecount/output_sort";				//实验输出目录
		Path inPath = new Path(hdfs + dataDir);
		Path outPath = new Path(hdfs + outputDir);
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outPath);
		FileSystem fs = FileSystem.get(conf);
		if(fs.exists(outPath)) {
			fs.delete(outPath, true);
		}
		
		//4.运行作业
		System.out.println("Job: " + jobName + " is running...");
		if(job.waitForCompletion(true)) {
			System.out.println("success!");
			System.exit(0);
		} else {
			System.out.println("failed!");
			System.exit(1);
		}
	}

}

<|start_filename|>src/main/java/mapreduceProgram/FlowSort.java<|end_filename|>
package mapreduceProgram;

import java.io.DataInput;
import java.io.DataOutput;
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.Text;
import org.apache.hadoop.io.WritableComparable;
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 FlowSort {

	public static class MySortKey implements WritableComparable<MySortKey> {
		private int upFlow;
		private int downFlow;
		private int sumFlow;

		public MySortKey() {
			// TODO Auto-generated constructor stub
		}
		
		public MySortKey(int up, int down) {
			upFlow = up;
			downFlow = down;
			sumFlow = up + down;
		}

		public int getUpFlow() {
			return upFlow;
		}

		public void setUpFlow(int upFlow) {
			this.upFlow = upFlow;
		}

		public int getDownFlow() {
			return downFlow;
		}

		public void setDownFlow(int downFlow) {
			this.downFlow = downFlow;
		}

		public int getSumFlow() {
			return sumFlow;
		}

		public void setSumFlow(int sumFlow) {
			this.sumFlow = sumFlow;
		}

		@Override
		public void write(DataOutput out) throws IOException {
			// TODO Auto-generated method stub
			out.writeInt(upFlow);
			out.writeInt(downFlow);
			out.writeInt(sumFlow);
		}

		@Override
		public void readFields(DataInput in) throws IOException {
			// TODO Auto-generated method stub
			upFlow = in.readInt();
			downFlow = in.readInt();
			sumFlow = in.readInt();
		}

		@Override
		public int compareTo(MySortKey o) {
			if ((this.upFlow - o.upFlow) == 0) {// 上行流量相等，比较下行流量
				return o.downFlow - this.downFlow;// 按downFlow降序排序
			} else {
				return this.upFlow - o.upFlow;// 按upFlow升序排
			}
		}

		@Override
		public String toString() {
			// TODO Auto-generated method stub
			return upFlow + "\t" + downFlow + "\t" + sumFlow;
		}
	}

	public static class SortMapper extends Mapper<Object, Text, MySortKey, Text> {
		Text phone = new Text();
		MySortKey mySortKey = new MySortKey();

		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
			String[] lists = value.toString().split("\t");
			phone.set(lists[0]);
			mySortKey.setUpFlow(Integer.parseInt(lists[1]));
			mySortKey.setDownFlow(Integer.parseInt(lists[2]));
			context.write(mySortKey, phone);
			System.out.println(phone.toString()+":"+mySortKey.toString()+",up:"+lists[1]+"=="+mySortKey.getUpFlow());
		}
	}

	public static class SortReducer extends Reducer<MySortKey, Text, Text, MySortKey> {
		public void reduce(MySortKey key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
			for (Text value : values) {
				System.out.println(value.toString()+","+key.toString());
				context.write(value, key);
			}
		}
	}

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		// 设置HDFS配置信息
		String namenode_ip = "192.168.17.10";
		String hdfs = "hdfs://" + namenode_ip + ":9000";
		Configuration conf = new Configuration();
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true");

		// 设置job配置信息
		Job job = Job.getInstance(conf, "FlowSort");
		job.setJarByClass(FlowSort.class);
		job.setJar("export\\FlowSort.jar");
		// Mapper
		job.setMapperClass(SortMapper.class);
		job.setMapOutputKeyClass(MySortKey.class);
		job.setMapOutputValueClass(Text.class);
		// Reducer
		job.setReducerClass(SortReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputKeyClass(MySortKey.class);
		// 作业输入输出路径
		String dataDir = "/workspace/flowStatistics/output/part-r-00000"; // 实验数据目录
		String outputDir = "/workspace/flowStatistics/output_sort"; // 实验输出目录
		Path inPath = new Path(hdfs + dataDir);
		Path outPath = new Path(hdfs + outputDir);
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outPath);
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(outPath)) {
			fs.delete(outPath, true);
		}
		// 运行作业
		System.out.println("Job: FlowSort is running...");
		if (job.waitForCompletion(true)) {
			System.out.println("success!");
			System.exit(0);
		} else {
			System.out.println("failed!");
			System.exit(1);
		}
	}
}


<|start_filename|>src/main/java/ssdut/training/mapreduce/medianstddev/MedianStdDevTuple.java<|end_filename|>
package ssdut.training.mapreduce.medianstddev;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.Writable;

public class MedianStdDevTuple implements Writable {
	private float median = 0f;
	private float stddev = 0f;

	public float getMedian() {
		return median;
	}

	public void setMedian(float median) {
		this.median = median;
	}

	public float getStddev() {
		return stddev;
	}

	public void setStddev(float stddev) {
		this.stddev = stddev;
	}

	@Override
	public void readFields(DataInput in) throws IOException {
		median = in.readFloat();
		stddev = in.readFloat();
	}

	@Override
	public void write(DataOutput out) throws IOException {
		out.writeFloat(median);
		out.writeFloat(stddev);
	}

	@Override
	public String toString() {
		return median + "\t" + stddev;
	}
	
}

<|start_filename|>src/main/java/mutualFriend/DecomposeFriendsMapper.java<|end_filename|>
package mutualFriend;

import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

import org.apache.hadoop.io.Text;

public class DecomposeFriendsMapper extends Mapper<Object, Text, Text, Text> {
	public void map(Object key,Text value,Context context) throws IOException, InterruptedException {
		String strs = value.toString();
		Text uString = new Text(strs.substring(0, 1));
		String[] friends = strs.substring(2).split(",");
		
		//A:B,C,D,F,E,O
		for (int i = 0; i < friends.length; i++) {
			// 以<B，A>,<C,A>形式输出
			context.write(new Text(friends[i]),uString);
		}
	}
}


<|start_filename|>src/main/java/ssdut/training/mapreduce/inputformat/FixedLengthInput2.java<|end_filename|>
package ssdut.training.mapreduce.inputformat;

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.BytesWritable;
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.input.FixedLengthInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class FixedLengthInput2 {
	public static class FixedLengthInput2Mapper extends Mapper<LongWritable, BytesWritable, Text, IntWritable> {		
		public void map(LongWritable key, BytesWritable value, Context context ) 
				throws IOException, InterruptedException {
			String val = new String(value.getBytes(), 0, value.getLength()-1);	
			String[] strs = val.split(" ");
			context.write(new Text(strs[0]), new IntWritable(Integer.parseInt(strs[1])));
	    }
	}
  
	public static class FixedLengthInput2Reducer extends Reducer<Text, IntWritable, Text, IntWritable> {
		public void reduce(Text key, Iterable<IntWritable> values, Context context) 
				throws IOException, InterruptedException {
			for (IntWritable val : values) {
				context.write(key, val);
			}
	    }
	}

	public static void main(String[] args) throws Exception {		
		//1.设置HDFS配置信息
		String namenode_ip = "192.168.17.10";
		String hdfs = "hdfs://" + namenode_ip + ":9000";			
		Configuration conf = new Configuration();
		conf.set("fs.defaultFS", hdfs);
		conf.set("mapreduce.app-submission.cross-platform", "true");
		conf.setInt(FixedLengthInputFormat.FIXED_RECORD_LENGTH, 13);
		
		//2.设置MapReduce作业配置信息
		String jobName = "FixedLengthInput2";					//作业名称
		Job job = Job.getInstance(conf, jobName);
		job.setJarByClass(FixedLengthInput2.class);				//指定运行时作业类
		job.setJar("export\\FixedLengthInput2.jar");				//指定本地jar包
		job.setMapperClass(FixedLengthInput2Mapper.class);		//指定Mapper类
		job.setMapOutputKeyClass(Text.class);			//设置Mapper输出Key类型
		job.setMapOutputValueClass(IntWritable.class);		//设置Mapper输出Value类型
		job.setReducerClass(FixedLengthInput2Reducer.class);		//指定Reducer类
		job.setOutputKeyClass(Text.class);				//设置Reduce输出Key类型
		job.setOutputValueClass(IntWritable.class); 			//设置Reduce输出Value类型
		
		job.setInputFormatClass(FixedLengthInputFormat.class);	//设置输入格式化类
		
		//3.设置作业输入和输出路径
		String dataDir = "/expr/fixedinput/data";			//实验数据目录	
		String outputDir = "/expr/fixedinput/output";		//实验输出目录
		Path inPath = new Path(hdfs + dataDir);
		Path outPath = new Path(hdfs + outputDir);
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outPath);
		FileSystem fs = FileSystem.get(conf);
		if(fs.exists(outPath)) {
			fs.delete(outPath, true);
		}
		
		//4.运行作业
		System.out.println("Job: " + jobName + " is running...");
		if(job.waitForCompletion(true)) {
			System.out.println("success!");
			System.exit(0);
		} else {
			System.out.println("failed!");
			System.exit(1);
		}
	}

}