计算每月温度最高的两天。
一、客户端
package zhazhalong.mr.tq;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class tqmr {
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration(true);
Job job = Job.getInstance(conf);
job.setJarByClass(tqmr.class);
//-----map start
job.setMapperClass(TMapper.class);
job.setMapOutputKeyClass(TQ.class);
job.setMapOutputValueClass(IntWritable.class);
//分区
job.setPartitionerClass(TPartitioner.class);
//排序比较器
job.setSortComparatorClass(TSortComparator.class);
//job.setCombinerClass(TCombiner.class);
//-----map end
//-----reduce start
//分组比较器：相同的key为一组
job.setGroupingComparatorClass(TGroupingCompartor.class);
job.setReducerClass(TReduce.class);
//reduce end
Path input =new Path("/root/tq.txt");
FileInputFormat.addInputPath(job, input );
Path output=new Path("/output/05");
FileOutputFormat.setOutputPath(job, output);
job.waitForCompletion(true);
}
}
二、Mapper
package zhazhalong.mr.tq;
import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.util.StringUtils;
public class TMapper extends Mapper<LongWritable, Text, TQ, IntWritable>{
//map工作就是做映射，对数据进行整理kv给reduce.
TQ mkey=new TQ();
IntWritable mval=new IntWritable();
@Override
protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, TQ, IntWritable>.Context context)
throws IOException, InterruptedException {
/*
* 1949-10-01 14:21:02 34c
*/
try {
String[] str=StringUtils.split(value.toString(), '\t');
SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd");
Date date=sdf.parse(str[0]);
Calendar cal=Calendar.getInstance();
cal.setTime(date);
mkey.setYear(cal.get(Calendar.YEAR));
mkey.setMonth(cal.get(Calendar.MONTH)+1);
mkey.setDay(cal.get(Calendar.DAY_OF_MONTH));
int wd=Integer.parseInt(str[1].substring(0, str[1].length()-1));//取出温度
mkey.setWd(wd);
mval.set(wd);
context.write(mkey, mval);
} catch (ParseException e) {
e.printStackTrace();
}
}
}
三、reduce
package zhazhalong.mr.tq;
import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class TReduce extends Reducer<TQ, IntWritable, Text, IntWritable>{
Text rkey=new Text();
IntWritable rval=new IntWritable();
@Override
protected void reduce(TQ key, Iterable<IntWritable> values,Context context)
throws IOException, InterruptedException {
/* 相同的key为一组
* key value
* 1970 01 01 88 88 //第一条最高
* 1970 01 11 78 78
* 1970 01 21 68 68
* 1970 01 01 58 58
*/
int flg=0;
int day=0;
for (IntWritable v:values) {
if(flg==0) {//第一高
rkey.set(key.getYear()+"-"+key.getMonth()+"-"+key.getDay());
//1970-01-01:88
rval.set(key.getWd());
flg++;
day=key.getDay();
context.write(rkey, rval);
}
if(flg!=0 && day!=key.getDay()) {
//第二高
rkey.set(key.getYear()+"-"+key.getMonth()+"-"+key.getDay()+":"+key.getWd());
//1970-01-01:88
rval.set(key.getWd());
context.write(rkey, rval);
break;
}
}
}
}
四、天气类
package zhazhalong.mr.tq;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.WritableComparable;
public class TQ implements WritableComparable<TQ>{
private int year;
private int month;
private int day;
private int wd;
public int getYear() {
return year;
}
public void setYear(int year) {
this.year = year;
}
public int getMonth() {
return month;
}
public void setMonth(int month) {
this.month = month;
}
public int getDay() {
return day;
}
public void setDay(int day) {
this.day = day;
}
public int getWd() {
return wd;
}
public void setWd(int wd) {
this.wd = wd;
}
//序列化成字节数组
@Override
public void write(DataOutput out) throws IOException {
out.writeInt(year);
out.writeInt(month);
out.writeInt(day);
out.writeInt(wd);
}
//反序列化
@Override
public void readFields(DataInput in) throws IOException {
this.year=in.readInt();
this.month=in.readInt();
this.day=in.readInt();
this.wd=in.readInt();
}
//比较器
@Override
public int compareTo(TQ that) {
//根据自己的定义：日期正序
int c1=Integer.compare(this.year, that.getYear());//比较
if(c1==0)//year相同
{
int c2=Integer.compare(this.month, that.getMonth());
if(c2==0)//month相同
{
return Integer.compare(this.day, that.getDay());
}
return c2;//按月排序
}
return c1;//按年排序
}
}
五、分区
package zhazhalong.mr.tq;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Partitioner;
public class TPartitioner extends Partitioner<TQ, IntWritable> {
@Override
public int getPartition(TQ key, IntWritable value, int numPartitions) {
//对key分组
//map输出的每一个kv都会调用这个方法
return key.getYear() % numPartitions;
}
}
六、排序比较器
package zhazhalong.mr.tq;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
public class TSortComparator extends WritableComparator{
public TSortComparator() {
super(TQ.class,true);
}
@Override
public int compare(WritableComparable a, WritableComparable b) {
TQ t1=(TQ)a;
TQ t2=(TQ)b;
int c1=Integer.compare(t1.getYear(), t2.getYear());
if(c1==0) {
int c2=Integer.compare(t1.getMonth(), t2.getMonth());
if(c2==0) {
return -Integer.compare(t1.getWd(), t2.getWd());//倒序
}
return c2;
}
return c1;
}
}
七、分组比较器
package zhazhalong.mr.tq;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
public class TGroupingCompartor extends WritableComparator{
public TGroupingCompartor() {
super(TQ.class,true);
}
@Override
public int compare(WritableComparable a, WritableComparable b) {
TQ t1=(TQ)a;
TQ t2=(TQ)b;
int c1=Integer.compare(t1.getYear(), t2.getYear());
if(c1==0) {
int c2=Integer.compare(t1.getMonth(), t2.getMonth());
return c2;
}
return c1;
}
}
good厉害了
看了那么多博客，就你的能看懂
挺明白的
好