Hadoop IO 特性详解（2）

FSDataInputStream/FSDataOutputStream

publicclassFSDataInputStreamextendsDataInputStream
implementsSeekable,PositionedReadable,Closeable,HasFileDescriptor{
publicFSDataInputStream(InputStreamin)throwsIOException{
super(in);
if(!(ininstanceofSeekable)||!(ininstanceofPositionedReadable)){
thrownewIllegalArgumentException("InisnotaninstanceofSeekableorPositionedReadable");
}
}

publicsynchronizedvoidseek(longdesired)throwsIOException{
((Seekable)in).seek(desired);
}

publicvoidreadFully(longposition,byte[]buffer)
throwsIOException{
((PositionedReadable)in).readFully(position,buffer,0,buffer.length);
}
...
}

publicinterfaceSeekable{
//Seektothegivenoffsetfromthestartofthefile.
voidseek(longpos)throwsIOException;

//Returnthecurrentoffsetfromthestartofthefile
longgetPos()throwsIOException;

//Seeksadifferentcopyofthedata.Returnstrueiffoundanewsource,falseotherwise.
booleanseekToNewSource(longtargetPos)throwsIOException;
}

publicinterfacePositionedReadable{
publicintread(longposition,byte[]buffer,intoffset,intlength)throwsIOException;

publicvoidreadFully(longposition,byte[]buffer,intoffset,intlength)throwsIOException;

publicvoidreadFully(longposition,byte[]buffer)throwsIOException;
}

publicclassFSDataOutputStreamextendsDataOutputStreamimplementsSyncable{...}

Hadoop具体文件系统

FileSystem<--FilterFileSystem<--ChecksumFileSystem<--LocalFileSystem
<--ChecksumDistributeFileSystem

ChecksumFileSystem

文件/目录相关事务

publicbooleandelete(Pathf,booleanrecursive)throwsIOException{
FileStatusfstatus=null;
try{
fstatus=fs.getFileStatus(f);
}catch(FileNotFoundExceptione){
returnfalse;
}
if(fstatus.isDir()){
returnfs.delete(f,recursive);
}else{
PathcheckFile=getChecksumFile(f);
if(fs.exists(checkFile)){
fs.delete(checkFile,true);
}
returnfs.delete(f,true);
}
}

读文件

abstractpublicclassFSInputCheckerextendsFSInputStream{
protectedPathfile;//Thefilenamefromwhichdataisreadfrom
privateChecksumsum;
privatebooleanverifyChecksum=true;
privatebyte[]buf;//数据缓冲区
privatebyte[]checksum;//校验和缓冲区
privateintpos;
privateintcount;
privateintnumOfRetries;//出错重试次数
privatelongchunkPos=0;//cachedfileposition
...
}

publicChecksumFSInputChecker(ChecksumFileSystemfs,Pathfile,intbufferSize)
throwsIOException{
super(file,fs.getFileStatus(file).getReplication());
...
try{
...
if(!Arrays.equals(version,CHECKSUM_VERSION))
thrownewIOException("Notachecksumfile:"+sumFile);
this.bytesPerSum=sums.readInt();
set(fs.verifyChecksum,newPureJavaCrc32(),bytesPerSum,4);
}catch(...){//ignore
set(fs.verifyChecksum,null,1,0);
}
}

publicsynchronizedintread(byte[]b,intoff,intlen)throwsIOException{
...//参数校验
intn=0;
for(;;){
intnread=read1(b,off+n,len-n);
if(nread<=0)
return(n==0)?nread:n;
n+=nread;
if(n>=len)
returnn;
}
}

privateintread1(byteb[],intoff,intlen)throwsIOException{
intavail=count-pos;
if(avail<=0){
if(len>=buf.length){
intnread=readChecksumChunk(b,off,len);//readachunktouserbufferdirectly;avoidonecopy
returnnread;
}else{
fill();//readachunkintothelocalbuffer
if(count<=0){
return-1;
}else{
avail=count;
}
}
}
//copycontentofthelocalbuffertotheuserbuffer
intcnt=(avail<len)?avail:len;
System.arraycopy(buf,pos,b,off,cnt);
pos+=cnt;
returncnt;
}

privateintreadChecksumChunk(byteb[],intoff,intlen)throwsIOException{
//invalidatebuffer
count=pos=0;
intread=0;
booleanretry=true;
intretriesLeft=numOfRetries;
do{
retriesLeft--;
try{
read=readChunk(chunkPos,b,off,len,checksum);
if(read>0){
if(needChecksum()){
sum.update(b,off,read);
verifySum(chunkPos);
}
chunkPos+=read;
}
retry=false;
}catch(ChecksumExceptionce){
if(retriesLeft==0){
throwce;
}
if(seekToNewSource(chunkPos)){//重试一个新的数据副本
seek(chunkPos);
}else{
throwce;
}
}
}while(retry);
returnread;
}

//ChecksumFSInputChecker.readChunk()
protectedintreadChunk(longpos,byte[]buf,intoffset,intlen,
byte[]checksum)throwsIOException{
booleaneof=false;
if(needChecksum()){
try{
longchecksumPos=getChecksumFilePos(pos);
if(checksumPos!=sums.getPos()){
sums.seek(checksumPos);
}
sums.readFully(checksum);
}catch(EOFExceptione){
eof=true;
}
len=bytesPerSum;
}
if(pos!=datas.getPos()){
datas.seek(pos);
}
intnread=readFully(datas,buf,offset,len);
if(eof&&nread>0){
thrownewChecksumException("Checksumerror:"+file+"at"+pos,pos);
}
returnnread;
}

写文件

abstractpublicclassFSOutputSummerextendsOutputStream{
privateChecksumsum;//datachecksum计算校验和
privatebytebuf[];//internalbufferforstoringdatabeforeitischecksumed输出数据缓冲区
privatebytechecksum[];//internalbufferforstoringchecksum校验和缓冲区
privateintcount;//Thenumberofvalidbytesinthebuffer.已使用空间计数
...
}

//ChecksumFileSystem.CheckSumFSOutputSummer
privatestaticclassChecksumFSOutputSummerextendsFSOutputSummer{
privateFSDataOutputStreamdatas;
privateFSDataOutputStreamsums;

...
@Override
protectedvoidwriteChunk(byte[]b,intoffset,intlen,byte[]checksum)throwsIOException{
datas.write(b,offset,len);
sums.write(checksum);
}
}

publicsynchronizedvoidwrite(byteb[],intoff,intlen)throwsIOException{
...//参数校验
for(intn=0;n<len;n+=write1(b,off+n,len-n)){}
}

privateintwrite1(byteb[],intoff,intlen)throwsIOException{
if(count==0&&len>=buf.length){
finalintlength=buf.length;
sum.update(b,off,length);
writeChecksumChunk(b,off,length,false);
returnlength;
}

//copyuserdatatolocalbuffer
intbytesToCopy=buf.length-count;
bytesToCopy=(len<bytesToCopy)?len:bytesToCopy;
sum.update(b,off,bytesToCopy);
System.arraycopy(b,off,buf,count,bytesToCopy);
count+=bytesToCopy;
if(count==buf.length){//localbufferisfull
flushBuffer();
}
returnbytesToCopy;
}

privatevoidwriteChecksumChunk(byteb[],intoff,intlen,booleankeep)throwsIOException{
inttempChecksum=(int)sum.getValue();
if(!keep){
sum.reset();
}
int2byte(tempChecksum,checksum);//整数转字节数组
writeChunk(b,off,len,checksum);
}

publicChecksumFSOutputSummer(ChecksumFileSystemfs,Pathfile,booleanoverwrite,intbufferSize,
shortreplication,longblockSize,Progressableprogress)throwsIOException{
super(newPureJavaCrc32(),fs.getBytesPerSum(),4);
intbytesPerSum=fs.getBytesPerSum();
this.datas=fs.getRawFileSystem().create(file,overwrite,bufferSize,replication,blockSize,progress);
intsumBufferSize=fs.getSumBufferSize(bytesPerSum,bufferSize);
this.sums=fs.getRawFileSystem().create(fs.getChecksumFile(file),true,sumBufferSize,replication,blockSize);
sums.write(CHECKSUM_VERSION,0,CHECKSUM_VERSION.length);
sums.writeInt(bytesPerSum);
}