<file_sep>package searchFileByWord;

import org.apache.hadoop.conf.Configuration;
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.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import scorelist.Score;

public class SEARCHDriver {

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf, "JobName");
		job.setJarByClass(searchFileByWord.SEARCHDriver.class);
		// TODO: specify a mapper
		job.setMapperClass(SearchMapper.class);
		// TODO: specify a reducer
		job.setReducerClass(SearchReducer.class);

		// TODO: specify output types
		job.setOutputKeyClass(Score.class);
		job.setOutputValueClass(NullWritable.class);

		// TODO: specify input and output DIRECTORIES (not files)
		FileInputFormat.setInputPaths(job, new Path("hdfs://10.42.117.115:9000/mr/score"));
		FileOutputFormat.setOutputPath(job, new Path("hdfs://10.42.117.115:9000/scorelistResult"));

		if (!job.waitForCompletion(true))
			return;
	}

}
<file_sep>package sortProfit;

import org.apache.hadoop.conf.Configuration;
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.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import scorelist.Score;

public class SPDriver {

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf, "JobName");
		job.setJarByClass(sortProfit.SPDriver.class);
		// TODO: specify a mapper
		job.setMapperClass(SPMapper.class);
		// TODO: specify a reducer
		job.setReducerClass(SPReducer.class);

		// TODO: specify output types
//		job.setMapOutputKeyClass(Profit.class);
//		job.setMapOutputKeyClass(NullWritable.class);
		job.setOutputKeyClass(Profit.class);
		job.setOutputValueClass(NullWritable.class);

		// TODO: specify input and output DIRECTORIES (not files)
		FileInputFormat.setInputPaths(job, new Path("hdfs://10.42.117.115:9000/mr/profit.txt"));
		FileOutputFormat.setOutputPath(job, new Path("hdfs://10.42.117.115:9000/SPprofit111111-result"));

		if (!job.waitForCompletion(true))
			return;
	}

}
<file_sep>package charcter;

import java.io.IOException;

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

public class cRedducer extends Reducer<Text, LongWritable, Text, LongWritable> {
//注意：迭代器只能遍历一次
	public void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
		// process values
		long sum=0;
		for (LongWritable val : values) {
			sum+=val.get();

		}
		context.write(key, new LongWritable(sum));
	}

}
<file_sep>package profit;

import org.apache.hadoop.conf.Configuration;
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 profitDriver {

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf, "JobName");
		job.setJarByClass(profit.profitDriver.class);
		// TODO: specify a mapper
		job.setMapperClass(profitMapper.class);
		// TODO: specify a reducer
		job.setReducerClass(profitReducer.class);
//		设置分区类
		job.setPartitionerClass(profitPartioner.class);
//		根据分区数量设置reducetask数量
//		如果reducetask的数量为1分区无效
		job.setNumReduceTasks(3);

		// TODO: specify output types
		job.setMapOutputKeyClass(IntWritable.class);
		job.setMapOutputValueClass(Profit.class);
		
		job.setOutputKeyClass(IntWritable.class);
		job.setOutputValueClass(IntWritable.class);

		// TODO: specify input and output DIRECTORIES (not files)
		FileInputFormat.setInputPaths(job, new Path("hdfs://10.42.117.115:9000/mr/profit.txt"));
		FileOutputFormat.setOutputPath(job, new Path("hdfs://10.42.117.115:9000/profit-result"));

		if (!job.waitForCompletion(true))
			return;
	}

}
<file_sep>package scorelist.copy;

import java.io.IOException;

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

public class maxMapper extends Mapper<LongWritable, Text, Text, LongWritable> {

	public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
//		获取一行数据
      String line = value.toString();
//      
      String[] arr = line.split(" ");
      context.write(new Text(arr[0]), new LongWritable(Long.parseLong(arr[1])));
	}

}
<file_sep>package flow2bylocation;

import java.io.File;
import java.io.IOException;

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

public class flowMapper extends Mapper<LongWritable, Text, Text,Flowbean> {

	public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
//		获取一行数据
      String line = value.toString();
//      
      String[] arr = line.split(" ");
      Flowbean f=new Flowbean();
      f.setPhone(arr[0]);
      f.setAddr(arr[1]);
      f.setName(arr[2]);
      f.setFlow(Integer.parseInt(arr[3]));
      
      context.write(new Text(f.getName()), f);
	}

}
<file_sep>package charcter;

import java.io.IOException;

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

public class cMapper extends Mapper<LongWritable, Text, Text,LongWritable> {

	public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
//		获取一行数据
		String line = value.toString();
//		拆分字符
		char[] cs = line.toCharArray();
		for(char c:cs){
			if(c==' ')
				continue;
			
			context.write(new Text(c+""), new LongWritable(1));
		}

	}

}
<file_sep>package searchFileByWord;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
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 SearchMapper extends Mapper<LongWritable, Text, Text, Text> {

	public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
//		获取map对应的切片
		FileSplit split = (FileSplit) context.getInputSplit();
//		获取文件名
		String name = split.getPath().getName();
		String line = value.toString();
		String[] arr = line.split(" ");
		for(String a:arr){
			context.write(new Text(a),new Text(name));
		}
		
		
		

	}

}
<file_sep>package flow2bylocation;

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

public class flow2Partioner extends Partitioner<Text, Flowbean>{

	@Override
	public int getPartition(Text key, Flowbean value, int numPartions) {
		String addr = value.getAddr();
		switch (addr) {
		case "bj":
			return 0;
		case "sh":
			return 1;
		case "sz":
			return 2;
		default:
			return 3;
	}
	

}
}<file_sep>package scoresumonetxt;

import java.io.IOException;

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

import com.sun.org.apache.bcel.internal.generic.NEW;

public class sumMapper extends Mapper<LongWritable, Text, Text, Score> {

	public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
//		获取一行数据
      String line = value.toString();
//      
      String[] arr = line.split(" ");
      Score score=new Score();
      score.setName(arr[0]);
      score.setValue1(Integer.parseInt(arr[1]));
      score.setValue2(Integer.parseInt(arr[2]));
      score.setValue3(Integer.parseInt(arr[3]));
      context.write(new Text(score.getName()),score);
	}

}
<file_sep>package charcter;

import java.io.ByteArrayInputStream;
import java.io.DataInput;
import java.io.DataInputStream;
import java.io.IOException;

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

public class CharGroup extends WritableComparator{
	@Override
	public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
		DataInput input1=new DataInputStream(new ByteArrayInputStream(b1));
		Text key1=new Text();
		try {
			key1.readFields(input1);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		key1.set(b1,s1,l1);
		DataInput input2=new DataInputStream(new ByteArrayInputStream(b2));
		Text key2=new Text();
		try {
			key2.readFields(input2);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		key2.set(b2,s2,l2);
		char c1 = key1.toString().charAt(0);
		char c2 = key2.toString().charAt(0);
//		ÅÐ¶Ïc1ºÍc2ÊÇ·ñÎªÐ¡Ð´×ÖÄ¸
		if(c1>='a'&&c1<='z'&&c2>='a'&&c2<='z'){
			return 0;
			
		}
		else if(c1>='A'&&c1<='Z'&&c2>='A'&&c2<='Z'){
			return 0;
			
		}
		else return c1-c2;
	}

}
<file_sep>package scorelist;

import java.io.IOException;

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

public class maxReducer extends Reducer<Score, NullWritable, Score, NullWritable> {

	
	
	public void reduce(Score key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
//		LongWritable max=new LongWritable(0);
//		在进行遍历的时候，hadoop为了节省空间采用的是地址复用技术
//		key=bob
//		itarable：684 340 312 548
//      val.set(684);
//		val.set(340);
//		for (LongWritable val : values) {
//			if(val.get()>max.get()){
////				赋值的是地址
//				max=val;
//			}
		
		
		context.write(key, NullWritable.get());
	}

}
<file_sep>package score1sumbyMoth;

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

public class scorePartioner extends Partitioner<Text, Score>{

	@Override
	public int getPartition(Text key, Score value, int numPartions) {
		int month = value.getMonth();
		switch (month) {
		case 1:
			return 0;
		case 2:
			return 1;
		case 3:
			return 2;
		default:
			return 3;
	}
	

}
}