hadoop之用户定制

狐狸情

Hadoop提供了9中内置数据类型，分别为：
BooleanWritable
ByteWritable
IntWritable
LongWritable
FloatWritable
DoubleWritable
Text(使用UTF8格式存储的文本)
NullWritable（空值的时候使用）
当然，用户也可以自定义数据类型，自定义数据类型时，要实现Writable接口；如果自定义的数据作为key使用，还需要实现Comparable接口，在hadoop中，实现WritableComparable接口即可，WritableComparable继承自Writable和Comparable接口。
代码大概内容如下：
public class Point3D implements WritableComparable<Point3D>{
private float x,y,z;
public float getX(){return x;}
public float getY(){return y;}
public float getZ(){return z;}
public void readFields(DataInput in) throws IOException{
x=in.readFloat();
y=in.readFloat();
z=in.readFloat();

}
public void write(DataOutput out) throws IOException{
out.writeFloat(x);
out.writeFloat(y);
out.writeFloat(z);
}
public int compareTo(Point3D p){

return 1,0,-1;
}

}
学过java的人不难理解以上的代码。
当然，hadoop还允许定制其他的内容，比如定制输入输出的格式，hadoop提供了丰富的内置数据输入格式，最常用的输入格式是TextInputFormat和KeyValueTextInputFormat
TextInputFormat是系统默认的输入格式，它能够将输入的内容以<K,V>的形式输出，读入一行时，输出的K为该行在文本中的偏移量，输出的V为该行的内容。
KeyValueInputFormat是一个常用的输入格式，可以将一个按照<K,V>格式存放的文本文件逐行读出，并解析为<K,V>
每个InputFormat都有两个任务，将输入文件转化为split集合以及将split转化为键值对。
查看InputFormat的API,他有两个函数getSplits(JobContext context)和createRecordReader(InputSplit split, TaskAttemptContext context)
因此自定制输入格式自然得自定义RecordReader。
查看RecordReader的API,主要有initialize(InputSplit split, TaskAttemptContext context)，getCurrentKey()，getCurrentValue（）等待我们去实现。
下面以一个不考虑词频的倒排索引为例，进行说明。为了能更细粒度的记录单词在文档中出现时的行位置信息，这里假定文档文件的标示就是文件名，单词在文档中的行位置信息就是该行在文档中的偏移量，则代码如下：
import org.apache.hadoop.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.mapreduce.lib.*;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.*;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.util.*;
import java.util.*;
import java.io.*;
import java.net.*;
public class InvertedIndex {
public static class MyInputFormat extends FileInputFormat<Text,Text>{
MyRecordReader mrr=new MyRecordReader();
public RecordReader<Text,Text> createRecordReader(InputSplit split,TaskAttemptContext context){
try{
mrr.initialize(split, context);
}
catch(Exception e){
e.printStackTrace();
}
return mrr;
}
}
public static class MyRecordReader extends RecordReader<Text,Text>{
String fileName;
LineRecordReader lrr=new LineRecordReader();
public Text getCurrentKey(){
return new Text(fileName+"@"+lrr.getCurrentKey());
}
public Text getCurrentValue(){
return lrr.getCurrentValue();
}
public voidinitialize(InputSplit split, TaskAttemptContext context) throws IOException,InterruptedException{
lrr.initialize(split, context);
fileName=((FileSplit)split).getPath().toString();
}
public float getProgress(){
return lrr.getProgress();
}
public void close()   throws IOException{
lrr.close();
}
public boolean nextKeyValue() throws IOException {
return lrr.nextKeyValue();
}
}

public static void main(String [] args) throws Exception
{
Configuration conf = new Configuration();
String[] otherArgs=new GenericOptionsParser(conf,args).getRemainingArgs();
if(otherArgs.length!=2){
System.err.println("error");
System.exit(1);
}
Job job=new Job(conf,"InvertedIndex");
    job.setJarByClass(InvertedIndex.class);

job.setInputFormatClass(MyInputFormat.class);

job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);

FileInputFormat.addInputPath(job,new Path(otherArgs[0]));
FileOutputFormat.setOutputPath(job,new Path(otherArgs[1]));
   
System.exit(job.waitForCompletion(true)?0:1);
      
}
}

上面的例子仅仅是为了如何演示自定义输入格式，实际上倒排索引程序完全没必要这样做。另外，这个例子是基于FileInputFormat和LineRecordReader实现的，当然我们也可以自己去继承RecordReader和InputFormat类，具体参见API即可。
以上就是hadoop中自定义内容的一些用法，可以自定义数据类型，自定义输入格式，当然也可以自定义输出格式，在此不再赘述。