题目：Data deduplication
描述
你的程序要求读入输入文件，在去掉所有数据中的重复数据后输出结果。在输入文件中每一行是一个元数据。
输入
输入是一组文本文件，在每个输入文件中每一行是一个数据。每一个元数据都是一个字符串。
输出文件
输出文件的每一行都是在输入文件中出现过的一个数据，并且输出文件中的每一行都不相同。
输入样例
input1:
2006-6-9 a
2006-6-10 b
2006-6-11 c
2006-6-12 d
2006-6-13 a
2006-6-14 b
2006-6-15 c
2006-6-11 c
input2:
2006-6-9 b
2006-6-10 a
2006-6-11 b
2006-6-12 d
2006-6-13 a
2006-6-14 c
2006-6-15 d
2006-6-11 c
输出样例:
2006-6-10 a
2006-6-10 b
2006-6-11 b
2006-6-11 c
2006-6-12 d
2006-6-13 a
2006-6-14 b
2006-6-14 c
2006-6-15 c
2006-6-15 d
2006-6-9 a
2006-6-9 b
注意:
1 输出结果是按照字典顺序排序的；
2 每一行都是一个元数据；
3 重复数据在输出文件中也要输出一次。
整整干了一下午加晚上，悲催的。。。。
程序应该有漏洞，只是在单机下小数据文件可以测试通过。欢迎指正~~
代码如下：
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
public class TextPair implements WritableComparable<TextPair> {
public Text first;
public Text second;
public TextPair(){
this.first=new Text();
this.second=new Text();
}
public TextPair(Text first, Text second) {
//super();
this.first = first;
this.second = second;
}
public TextPair(String first,String second){
this.first=new Text(first);
this.second=new Text(second);
}
public Text getFirst() {
return first;
}
public void setFirst(Text first) {
this.first = first;
}
public Text getSecond() {
return second;
}
public void setSecond(Text second) {
this.second = second;
}
public void set(Text first,Text second){
this.first=first;
this.second=second;
}
@Override
public int hashCode() {
// TODO Auto-generated method stub
return first.hashCode()*163+second.hashCode();
}
@Override
public boolean equals(Object obj) {
// TODO Auto-generated method stub
if(obj instanceof TextPair){
TextPair tp=(TextPair)obj;
return first.equals(tp.getFirst())&&second.equals(tp.getSecond());
}
return false;
}
@Override
public String toString() {
// TODO Auto-generated method stub
return first+"\t"+second;
}
@Override
public void readFields(DataInput in) throws IOException {
// TODO Auto-generated method stub
first.readFields(in);
second.readFields(in);
}
@Override
public void write(DataOutput out) throws IOException {
// TODO Auto-generated method stub
first.write(out);
second.write(out);
}
@Override
public int compareTo(TextPair tp) {
// TODO Auto-generated method stub
int cmp=first.compareTo(tp.getFirst());
if(cmp!=0)
return cmp;
return second.compareTo(tp.getSecond());
}
}
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;
public class FirstPartitioner extends Partitioner<TextPair, Text> {
@Override
public int getPartition(TextPair key, Text value, int numPartitions) {
// TODO Auto-generated method stub
return Math.abs(key.getFirst().hashCode()&Integer.MAX_VALUE)%numPartitions;
}
}
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
public class GroupComparator extends WritableComparator {
public GroupComparator() {
super(TextPair.class, true);
// TODO Auto-generated constructor stub
}
@Override
public int compare(WritableComparable a, WritableComparable b) {
// TODO Auto-generated method stub
TextPair p1=(TextPair)a;
TextPair p2=(TextPair)b;
return p1.getFirst().compareTo(p2.getFirst());
}
}
import java.io.IOException;
import java.util.StringTokenizer;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Mapper;
public class DatadeduplicationMapper extends Mapper<LongWritable,Text,TextPair,Text>{
@Override
protected void map(LongWritable key, Text value,
org.apache.hadoop.mapreduce.Mapper.Context context)
throws IOException, InterruptedException {
// TODO Auto-generated method stub
System.out.println("mapper");
String line=value.toString();
StringTokenizer token=new StringTokenizer(line);
String []str=new String[2];
int i=0;
TextPair tp=new TextPair();
while(token.hasMoreTokens()){
str[i]=token.nextToken();
i++;
}
tp.set(new Text(str[0]), new Text(str[1]));
System.out.println(tp);
System.out.println(str[1]);
context.write(tp, new Text(str[1]));
}
}
import java.awt.List;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.Vector;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class DatadeduplicationReducer extends
Reducer<TextPair, Text, Text, Text> {
@Override
protected void reduce(TextPair key, Iterable<Text> values,
Context context)
throws IOException, InterruptedException {
// TODO Auto-generated method stub
System.out.println("reducer");
Set set=new HashSet();
Vector<String>list=new Vector();
for(Text value:values){
//set.add(value.toString());
if(!list.contains(value.toString())){
list.add(value.toString());
}
System.out.println(key);
System.out.println(value);
}
for(int i=0;i<list.size();i++)
{
//System.out.println(it.next());
System.out.println((String)list.get(i));
context.write(key.getFirst(), new Text(list.get(i).toString()));
}
// context.write(key, values)
}
}
import java.io.IOException;
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 Datadeduplication {
/**
* @param args
* @throws IOException
* @throws ClassNotFoundException
* @throws InterruptedException
*/
public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
// TODO Auto-generated method stub
System.out.println("nihao");
Job job=new Job();
System.out.println("nihao");
job.setJobName("Datadeduplication");
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
job.setMapperClass(DatadeduplicationMapper.class);
job.setMapOutputKeyClass(TextPair.class);
job.setMapOutputValueClass(Text.class);
job.setPartitionerClass(FirstPartitioner.class);
job.setGroupingComparatorClass(GroupComparator.class);
//job.setSortComparatorClass(FirstGroupingComparator.class);
//job.setSortComparatorClass(GroupComparator.class);
//job.setGroupingComparatorClass(KeyComparator.class);
job.setReducerClass(DatadeduplicationReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
job.waitForCompletion(false);
}
}