Hadoop的MapReduce中多文件输出

2010-01-0811:142883人阅读评论(4)收藏举报inkfish原创，请勿商业性质转载，转载请注明来源(/inkfish)。p来测试结果。ServerbitJDK0_16-b01Writerviewplainpackageinkfish.hadoop.study;2.importjava.io.DataOutputStream;3.importjava.io.IOException;4.importjava.io.UnsupportedEncodingException;5.importorg.apache.hadoop.io.NullWritable;6.importorg.apache.hadoop.io.Text;7.importorg.apache.hadoop.mapreduce.RecordWriter;8.importorg.apache.hadoop.mapreduce.TaskAttemptContext;9.importorg.apache.hadoop.mapreduce.lib.output.TextOutputFormat;neRecordWriterKVextendsRecordWriterKV12.privatestaticfinalStringutf8="UTF-8";13.privatestaticfinalbyte[]newline;14.static{15.try{16.newline="/n".getBytes(utf8);17.}catch(UnsupportedEncodingExceptionuee){18.thrownewIllegalArgumentException("can'tfind"+utf8+"enco19.}20.}21.protectedDataOutputStreamout;22.privatefinalbyte[]keyValueSeparator;23.publicLineRecordWriter(DataOutputStreamout,StringkeyValueSeparator){24.this.out=out;25.try{26.this.keyValueSeparator=keyValueSeparator.getBytes(utf8);27.}catch(UnsupportedEncodingExceptionuee){28.thrownewIllegalArgumentException("can'tfind"+utf8+"enco29.}30.}31.publicLineRecordWriter(DataOutputStreamout){32.this(out,"/t");33.}34.privatevoidwriteObject(Objecto)throwsIOException{35.if(oinstanceofText){36.Textto=(Text)o;37.out.write(to.getBytes(),0,to.getLength());38.}else{39.out.write(o.toString().getBytes(utf8));40.}41.}42.publicsynchronizedvoidwrite(Kkey,Vvalue)throwsIOException{43.booleannullKey=key==null||keyinstanceofNullWritable;44.booleannullValue=value==null||valueinstanceofNullWritable;45.if(nullKey&&nullValue){46.return;47.}48.if(!nullKey){49.writeObject(key);50.}51.if(!(nullKey||nullValue)){52.out.write(keyValueSeparator);53.}54.if(!nullValue){55.writeObject(value);56.}57.out.write(newline);58.}59.publicsynchronizedvoidclose(TaskAttemptContextcontext)throwsIOException{60.out.close();61.}ltipleOutputFormatconf)，即通过key和value及conf配置信息决定文件名(含扩展名)。viewplainpackageinkfish.hadoop.study;importjava.io.DataOutputStream;.importjava.io.IOException;4.importjava.util.HashMap;importjava.util.Iterator;importorgapache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.FSDataOutputStream;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.io.Writable;CodecmmitterdWritermptContextputFileOutputCommittertputFileOutputFormationUtilsutFormatKextendsWritableComparableVextendsWritable>20.extendsFileOutputFormat<K,V>{21.privateMultiRecordWriterwriter=null;22.publicRecordWriter<K,V>getRecordWriter(TaskAttemptContextjob)throws23.InterruptedException{24.if(writer==null){25.writer=newMultiRecordWriter(job,getTaskOutputPath(job));26.}27.returnwriter;28.}29.privatePathgetTaskOutputPath(TaskAttemptContextconf)throwsIOException{30.PathworkPath=null;31.OutputCommittercommitter=super.getOutputCommitter(conf);32.if(committerinstanceofFileOutputCommitter){33.workPath=((FileOutputCommitter)committer).getWorkPath();34.}else{35.PathoutputPath=super.getOutputPath(conf);36.if(outputPath==null){37.thrownewIOException("Undefinedjoboutput-path");38.}39.workPath=outputPath;40.}41.returnworkPath;42.}43./**通过key,value,conf来确定输出文件名(含扩展名)*/44.protectedabstractStringgenerateFileNameForKeyValue(Kkey,Vvalue,Cotionconf45.publicclassMultiRecordWriterextendsRecordWriter<K,V>{46./**RecordWriter的缓存*/47.privateHashMap<String,RecordWriter<K,V>>recordWriters=null;48.privateTaskAttemptContextjob=null;49./**输出目录*/50.privatePathworkPath=null;51.publicMultiRecordWriter(TaskAttemptContextjob,PathworkPath){52.super();53.this.job=job;54.this.workPath=workPath;55.recordWriters=newHashMap<String,RecordWriter<K,V>>();56.}57.@Override58.publicvoidclose(TaskAttemptContextcontext)throwsIOException,InterruptedException{59.Iterator<RecordWriter<K,V>>values=this.recordWriters.values(60.while(values.hasNext()){61.values.next().close(context);62.}63.this.recordWriters.clear();64.}65.@Override66.publicvoidwrite(Kkey,Vvalue)throwsIOException,InterruptedException{67.//得到输出文件名68.StringbaseName=generateFileNameForKeyValue(key,value,job.getConfiguration;69.RecordWriter<K,V>rw=this.recordWriters.get(baseName);70.if(rw==null){71.rw=getBaseRecordWriter(job,baseName);72.this.recordWriters.put(baseName,rw);73.}74.rw.write(key,value);75.}76.//${mapred.out.dir}/_temporary/_${taskid}/${nameWithExtension}77.privateRecordWriter<K,V>getBaseRecordWriter(TaskAttemptContextjongbaseName78.throwsIOException,InterruptedException{79.Configurationconf=job.getConfiguration();80.booleanisCompressed=getCompressOutput(job);81.StringkeyValueSeparator=",";82.RecordWriter<K,V>recordWriter=null;83.if(isCompressed){84.Class<?extendsCompressionCodec>codecClass=getOutputCompob85.GzipCodec.class);86.CompressionCodeccodec=ReflectionUtils.newInstance(codecClconf87.Pathfile=newPath(workPath,baseName+codec.getDefaultExtension));88.FSDataOutputStreamfileOut=file.getFileSystem(conf).create89.recordWriter=newLineRecordWriter<K,V>(newDataOutputStre90..createOutputStream(fileOut)),keyValueSeparator);91.}else{92.Pathfile=newPath(workPath,baseName);93.FSDataOutputStreamfileOut=file.getFileSystem(conf).create94.recordWriter=newLineRecordWriter<K,V>(fileOut,keyValueSrator95.}96.returnrecordWriter;97.}98.}ordCountttxt其他以“other.txt”保存。viewplainpackageinkfish.hadoop.study;.importjava.io.IOException;importjava.util.StringTokenizer;4.importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.io.IntWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Job;importorg.apache.hadoop.mapreduce.Mapper;ser15.publicstaticclassTokenizerMapperextendsMapper<Object,Text,Text,IntWritable>{privatefinalstaticIntWritableone=newIntWritable(1);17.privateTextword=newText();18.publicvoidmap(Objectkey,Textvalue,Contextcontext)throwsIOEx19.InterruptedException{20.StringTokenizeritr=newStringTokenizer(value.toString());21.while(itr.hasMoreTokens()){22.word.set(itr.nextToken());23.context.write(word,one);24.}25.}26.}27.publicstaticclassIntSumReducerextendsReducer<Text,IntWritable,Text,IntWritable>{28.privateIntWritableresult=newIntWritable();29.publicvoidreduce(Textkey,Iterable<IntWritable>values,Contextctext30.throwsIOException,InterruptedException{31.intsum=0;32.for(IntWritableval:values){33.sum+=val.get();34.}35.result.set(sum);36.context.write(key,result);37.}38.}39.publicstaticclassAlphabetOutputFormatextendsMultipleOutputFormat<Text,IntWritable>{40.@OtectedStringgenerateFileNameForKeyValue(Textkey,IntWritablevalue,Configurationconf){42.charc=key.toString().toLowerCase().charAt(0);43.if(c>='a'&&c<='z'){44.returnc+".txt";45.}46.return"other.txt";47.}48.}49.publicstaticvoidmain(String[]args)throwsException{50.Configurationconf=newConfiguration();51.String[]otherArgs=newGenericOptionsParser(conf,args).getRemaini52.if(otherArgs.length!=2){53.System.err.println("Usage:wordcount<in><out>");54.System.exit(2);67.}}bnewJobconfwordcountByClassWordCountclasserClassTokenizerMapperclassinerClassInt