Hadoop實(shí)驗(yàn)二-熟悉常用的HDFS操作

Hadoop實(shí)驗(yàn)?——熟悉常?的HDFS操作1.編程實(shí)現(xiàn)以下指定功能，并利?Hadoop提供的Shell命令完成相同任務(wù)：（1）向HDFS中上傳任意?本?件，如果指定的?件在HDFS中已經(jīng)存在，由?戶指定是追加到原有?件末尾還是覆蓋原有的?件；shell命令實(shí)現(xiàn)?先啟動所有的hadoop應(yīng)?上傳本地?件到HDFShadoopfs-puttext.txt/Test/追加到?件末尾的指令hadoopfs-appendToFilelocal.txt/Test/text.txt查看HDFS?件的內(nèi)容hadoopfs-cat/Test/text.txt覆蓋原有?件的指令hadoopfs-copyFromLocal-flocal.txt/Test/text.txt以上過程也可以?shell腳本實(shí)現(xiàn)：if$(hadoopfs-test-etext.txt);then$(hadoopfs-appendToFilelocal.txt/Test/text.txt);else$(hadoopfs-copyFromLocal-flocal.txt/Test/text.txt);fiJAVA實(shí)現(xiàn)importjava.io.FileInputStream;importjava.io.IOException;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.FSDataOutputStream;importorg.apache.hadoop.fs.FileSystem;importorg.apache.hadoop.fs.Path;publicclassCopyFromLocalFile{/***判斷路徑是否存在*/publicstaticbooleantest(Configurationconf,Stringpath){try(FileSystemfs=FileSystem.get(conf)){returnfs.exists(newPath(path));}catch(IOExceptione){e.printStackTrace();returnfalse;}}/***復(fù)制?件到指定路徑若路徑已存在，則進(jìn)?覆蓋*/publicstaticvoidcopyFromLocalFile(Configurationconf,StringlocalFilePath,StringremoteFilePath){PathlocalPath=newPath(localFilePath);PathremotePath=newPath(remoteFilePath);try(FileSystemfs=FileSystem.get(conf)){/*fs.copyFromLocalFile第?個參數(shù)表?是否刪除源?件，第?個參數(shù)表?是否覆蓋*/fs.copyFromLocalFile(false,true,localPath,remotePath);}catch(IOExceptione){e.printStackTrace();}}/***追加?件內(nèi)容*/publicstaticvoidappendToFile(Configurationconf,StringlocalFilePath,StringremoteFilePath){PathremotePath=newPath(remoteFilePath);try(FileSystemfs=FileSystem.get(conf);FileInputStreamin=newFileInputStream(localFilePath);){FSDataOutputStreamout=fs.append(remotePath);byte[]data=newbyte[1024];intread=-1;while((read=in.read(data))>0){out.write(data,0,read);}out.close();}catch(IOExceptione){e.printStackTrace();}}/***主函數(shù)*/publicstaticvoidmain(String[]args){Configurationconf=newConfiguration();conf.set("fs.defaultFS","hdfs://localhost:9000");StringlocalFilePath="/usr/local/hadoop/text.txt";//本地路徑StringlocalFilePath="/usr/local/hadoop/text.txt";//本地路徑StringremoteFilePath="/user/tiny/text.txt";//HDFS路徑//Stringchoice="append";//若?件存在則追加到?件末尾Stringchoice="overwrite";//若?件存在則覆蓋try{/*判斷?件是否存在*/booleanfileExists=false;if(CopyFromLocalFile.test(conf,remoteFilePath)){fileExists=true;System.out.println(remoteFilePath+"已存在.");}else{System.out.println(remoteFilePath+"不存在.");}/*進(jìn)?處理*/if(!fileExists){//?件不存在，則上傳CopyFromLocalFile.copyFromLocalFile(conf,localFilePath,remoteFilePath);System.out.println(localFilePath+"已上傳?"+remoteFilePath);}elseif(choice.equals("overwrite")){//選擇覆蓋CopyFromLocalFile.copyFromLocalFile(conf,localFilePath,remoteFilePath);System.out.println(localFilePath+"已覆蓋"+remoteFilePath);}elseif(choice.equals("append")){//選擇追加CopyFromLocalFile.appendToFile(conf,localFilePath,remoteFilePath);System.out.println(localFilePath+"已追加?"+remoteFilePath);}}catch(Exceptione){e.printStackTrace();}}}（2）從HDFS中下載指定?件，如果本地?件與要下載的?件名稱相同，則?動對下載的?件重命名；Shell命令實(shí)現(xiàn)if$(hadoopfs-test-e/usr/local/hadoop/text.txt);then$(hadoopfs-copyToLocal/Test/text.txt./text.txt);else$(hadoopfs-copyToLocal/Test/text.txt./text2.txt);fiJAVA實(shí)現(xiàn)：importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.*;importorg.apache.hadoop.fs.FileSystem;importjava.io.*;publicclassCopyToLocal{/***下載?件到本地判斷本地路徑是否已存在，若已存在，則?動進(jìn)?重命名*/publicstaticvoidcopyToLocal(Configurationconf,StringremoteFilePath,StringlocalFilePath){PathremotePath=newPath(remoteFilePath);try(FileSystemfs=FileSystem.get(conf)){Filef=newFile(localFilePath);/*如果?件名存在，?動重命名(在?件名后?加上_0,_1...)*/if(f.exists()){System.out.println(localFilePath+"已存在.");Integeri=Integer.valueOf(0);while(true){f=newFile(localFilePath+"_"+i.toString());if(!f.exists()){localFilePath=localFilePath+"_"+i.toString();break;}else{i++;continue;}}System.out.println("將重新命名為:"+localFilePath);}//下載?件到本地PathlocalPath=newPath(localFilePath);fs.copyToLocalFile(remotePath,localPath);}catch(IOExceptione){//TODOAuto-generatedcatchblocke.printStackTrace();}}/***主函數(shù)*/publicstaticvoidmain(String[]args){Configurationconf=newConfiguration();conf.set("fs.defaultFS","hdfs://localhost:9000");StringlocalFilePath="/usr/local/hadoop/text.txt";//本地路徑StringremoteFilePath="/user/tiny/text.txt";//HDFS路徑try{CopyToLocal.copyToLocal(conf,remoteFilePath,localFilePath);System.out.println("下載完成");}catch(Exceptione){e.printStackTrace();}}}（3）將HDFS中指定?件的內(nèi)容輸出到終端中；Shell腳本實(shí)現(xiàn)：hadoopfs-cat/Test/text.txtJAVA實(shí)現(xiàn)importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.*;importorg.apache.hadoop.fs.FileSystem;importjava.io.*;publicclassCat{/***讀取?件內(nèi)容*/publicstaticvoidcat(Configurationconf,StringremoteFilePath){PathremotePath=newPath(remoteFilePath);try(FileSystemfs=FileSystem.get(conf);FSDataInputStreamin=fs.open(remotePath);BufferedReaderd=newBufferedReader(newInputStreamReader(in));){Stringline;while((line=d.readLine())!=null){System.out.println(line);}}catch(IOExceptione){e.printStackTrace();}}/***主函數(shù)*/publicstaticvoidmain(String[]args){Configurationconf=newConfiguration();conf.set("fs.defaultFS","hdfs://localhost:9000");StringremoteFilePath="/user/tiny/input/text.txt";//HDFS路徑try{System.out.println("讀取?件:"+remoteFilePath);Cat.cat(conf,remoteFilePath);System.out.println("\n讀取完成");}catch(Exceptione){e.printStackTrace();}}}（4）顯?HDFS中指定的?件的讀寫權(quán)限、??、創(chuàng)建時間、路徑等信息；Shell命令實(shí)現(xiàn)hadoopfs-ls-h/Test/text.txtJAVA實(shí)現(xiàn)：importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.*;importorg.apache.hadoop.fs.FileSystem;importjava.io.*;importjava.text.SimpleDateFormat;publicclassList{/***顯?指定?件的信息*/publicstaticvoidls(Configurationconf,StringremoteFilePath){try(FileSystemfs=FileSystem.get(conf)){PathremotePath=newPath(remoteFilePath);FileStatus[]fileStatuses=fs.listStatus(remotePath);for(FileStatuss:fileStatuses){System.out.println("路徑:"+s.getPath().toString());System.out.println("權(quán)限:"+s.getPermission().toString());System.out.println("??:"+s.getLen());/*返回的是時間戳,轉(zhuǎn)化為時間?期格式*/longtimeStamp=s.getModificationTime();SimpleDateFormatformat=newSimpleDateFormat("yyyy-MM-ddHH:mm:ss");Stringdate=format.format(timeStamp);System.out.println("時間:"+date);}}catch(IOExceptione){e.printStackTrace();}}/***主函數(shù)*/publicstaticvoidmain(String[]args){Configurationconf=newConfiguration();conf.set("fs.defaultFS","hdfs://localhost:9000");StringremoteFilePath="/user/tiny/text.txt";//HDFS路徑try{System.out.println("讀取?件信息:"+remoteFilePath);List.ls(conf,remoteFilePath);System.out.println("\n讀取完成");}catch(Exceptione){e.printStackTrace();}}}（5）給定HDFS中某?個?錄，輸出該?錄下的所有?件的讀寫權(quán)限、??、創(chuàng)建時間、路徑等信息，如果該?件是?錄，則遞歸輸出該?錄下所有?件相關(guān)信息；Shell命令實(shí)現(xiàn)：hadoopfs-ls-R-h/TestJAVA實(shí)現(xiàn)：importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.*;importorg.apache.hadoop.fs.FileSystem;importjava.io.*;importjava.text.SimpleDateFormat;publicclassListDir{/***顯?指定?件夾下所有?件的信息（遞歸）*/publicstaticvoidlsDir(Configurationconf,StringremoteDir){try(FileSystemfs=FileSystem.get(conf)){PathdirPath=newPath(remoteDir);/*遞歸獲取?錄下的所有?件*/RemoteIterator<LocatedFileStatus>remoteIterator=fs.listFiles(dirPath,true);/*輸出每個?件的信息*/while(remoteIterator.hasNext()){FileStatuss=remoteIterator.next();System.out.println("路徑:"+s.getPath().toString());System.out.println("權(quán)限:"+s.getPermission().toString());System.out.println("??:"+s.getLen());/*返回的是時間戳,轉(zhuǎn)化為時間?期格式*/LongtimeStamp=s.getModificationTime();SimpleDateFormatformat=newSimpleDateFormat("yyyy-MM-ddHH:mm:ss");Stringdate=format.format(timeStamp);System.out.println("時間:"+date);System.out.println();}}catch(IOExceptione){e.printStackTrace();}}/***主函數(shù)*/publicstaticvoidmain(String[]args){Configurationconf=newConfiguration();conf.set("fs.defaultFS","hdfs://localhost:9000");StringremoteDir="/user/tiny";//HDFS路徑try{System.out.println("(遞歸)讀取?錄下所有?件的信息:"+remoteDir);ListDir.lsDir(conf,remoteDir);System.out.println("讀取完成");}catch(Exceptione){e.printStackTrace();}}}（6）提供?個HDFS內(nèi)的?件的路徑，對該?件進(jìn)?創(chuàng)建和刪除操作。如果?件所在?錄不存在，則?動創(chuàng)建?錄；Shell腳本實(shí)現(xiàn)：#!/bin/bashif$(hadoopfs-test-d/Test/test1);then$(hadoopfs-touchz/Test/test1/text1.txt);else$(hadoopfs-mkdir-p/Test/test1&&hdfsdfs-touchz/Test/test1/text1.txt);fiJAVA實(shí)現(xiàn)：importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.*;importjava.io.*;publicclassRemoveOrMake{/***判斷路徑是否存在*/publicstaticbooleantest(Configurationconf,Stringpath){try(FileSystemfs=FileSystem.get(conf)){returnfs.exists(newPath(path));}catch(IOExceptione){e.printStackTrace();returnfalse;}}/***創(chuàng)建?錄*/publicstaticbooleanmkdir(Configurationconf,StringremoteDir){try(FileSystemfs=FileSystem.get(conf)){PathdirPath=newPath(remoteDir);returnfs.mkdirs(dirPath);}catch(IOExceptione){e.printStackTrace();returnfalse;}}/***創(chuàng)建?件*/publicstaticvoidtouchz(Configurationconf,StringremoteFilePath){PathremotePath=newPath(remoteFilePath);try(FileSystemfs=FileSystem.get(conf)){FSDataOutputStreamoutputStream=fs.create(remotePath);outputStream.close();}catch(IOExceptione){e.printStackTrace();}}/***刪除?件*/publicstaticbooleanrm(Configurationconf,StringremoteFilePath){PathremotePath=newPath(remoteFilePath);try(FileSystemfs=FileSystem.get(conf)){returnfs.delete(remotePath,false);}catch(IOExceptione){e.printStackTrace();e.printStackTrace();returnfalse;}}/***主函數(shù)*/publicstaticvoidmain(String[]args){Configurationconf=newConfiguration();conf.set("fs.defaultFS","hdfs://localhost:9000");StringremoteFilePath="/user/tiny/input/text.txt";//HDFS路徑StringremoteDir="/user/tiny/input";//HDFS路徑對應(yīng)的?錄try{/*判斷路徑是否存在，存在則刪除，否則進(jìn)?創(chuàng)建*/if(RemoveOrMake.test(conf,remoteFilePath)){RemoveOrMake.rm(conf,remoteFilePath);//刪除System.out.println("刪除?件:"+remoteFilePath);}else{if(!RemoveOrMake.test(conf,remoteDir)){//若?錄不存在，則進(jìn)?創(chuàng)建RemoveOrMake.mkdir(conf,remoteDir);System.out.println("創(chuàng)建?件夾:"+remoteDir);}RemoveOrMake.touchz(conf,remoteFilePath);System.out.println("創(chuàng)建?件:"+remoteFilePath);}}catch(Exceptione){e.printStackTrace();}}}（8）向HDFS中指定的?件追加內(nèi)容，由?戶指定內(nèi)容追加到原有?件的開頭或結(jié)尾；Shell命令實(shí)現(xiàn)：追加到?件結(jié)尾hadoopfs-appendToFilelocal.txt/Test/text.txt追加到?件開頭hadoopfs-get/Test/text.txtcat/Test/text.txt>>local.txthadoopfs-copyFromLocal-ftext.txt/Test/text.txtJAVA實(shí)現(xiàn)：importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.*;importorg.apache.hadoop.fs.FileSystem;importjava.io.*;publicclassAppendToFile{/***判斷路徑是否存在*/publicstaticbooleantest(Configurationconf,Stringpath){try(FileSystemfs=FileSystem.get(conf)){returnfs.exists(newPath(path));}catch(IOExceptione){e.printStackTrace();returnfalse;}}/***追加?本內(nèi)容*/publicstaticvoidappendContentToFile(Configurationconf,Stringcontent,StringremoteFilePath){try(FileSystemfs=FileSystem.get(conf)){PathremotePath=newPath(remoteFilePath);/*創(chuàng)建?個?件輸出流，輸出的內(nèi)容將追加到?件末尾*/FSDataOutputStreamout=fs.append(remotePath);out.write(content.getBytes());out.close();}catch(IOExceptione){e.printStackTrace();}}/***追加?件內(nèi)容*/publicstaticvoidappendToFile(Configurationconf,StringlocalFilePath,StringremoteFilePath){PathremotePath=newPath(remoteFilePath);try(FileSystemfs=FileSystem.get(conf);FileInputStreamin=newFileInputStream(localFilePath);){FSDataOutputStreamout=fs.append(remotePath);byte[]data=newbyte[1024];intread=-1;while((read=in.read(data))>0){while((read=in.read(data))>0){out.write(data,0,read);}out.close();}catch(IOExceptione){e.printStackTrace();}}/***移動?件到本地移動后，刪除源?件*/publicstaticvoidmoveToLocalFile(Configurationconf,StringremoteFilePath,StringlocalFilePath){try(FileSystemfs=FileSystem.get(conf)){PathremotePath=newPath(remoteFilePath);PathlocalPath=newPath(localFilePath);fs.moveToLocalFile(remotePath,localPath);}catch(IOExceptione){e.printStackTrace();}}/***創(chuàng)建?件*/publicstaticvoidtouchz(Configurationconf,StringremoteFilePath){try(FileSystemfs=FileSystem.get(conf)){PathremotePath=newPath(remoteFilePath);FSDataOutputStreamoutputStream=fs.create(remotePath);outputStream.close();}catch(IOExceptione){e.printStackTrace();}}/***主函數(shù)*/publicstaticvoidmain(String[]args){Configurationconf=newConfiguration();conf.set("fs.defaultFS","hdfs://localhost:9000");StringremoteFilePath="/user/tiny/text.txt";//HDFS?件Stringcontent="新追加的內(nèi)容\n";Stringchoice="after";//追加到?件末尾//Stringchoice="before";//追加到?件開頭try{/*判斷?件是否存在*/if(!AppendToFile.test(conf,remoteFilePath)){System.out.println("?件不存在:"+remoteFilePath);}else{if(choice.equals("after")){//追加在?件末尾AppendToFile.appendContentToFile(conf,content,remoteFilePath);System.out.println("已追加內(nèi)容到?件末尾"+remoteFilePath);}elseif(choice.equals("before")){//追加到?件開頭/*沒有相應(yīng)的api可以直接操作，因此先把?件移動到本地，創(chuàng)建?個新的HDFS，再按順序追加內(nèi)容*/StringlocalTmpPath="/user/hadoop/tmp.txt";AppendToFile.moveToLocalFile(conf,remoteFilePath,localTmpPath);//移動到本地AppendToFile.touchz(conf,remoteFilePath);//創(chuàng)建?個新?件AppendToFile.appendContentToFile(conf,content,remoteFilePath);//先寫?新內(nèi)容AppendToFile.appendToFile(conf,localTmpPath,remoteFilePath);//再寫?原來內(nèi)容remoteFilePath);//再寫?原來內(nèi)容System.out.println("已追加內(nèi)容到?件開頭:"+remoteFilePath);}}}catch(Exceptione){e.printStackTrace();}}}（9）刪除HDFS中指定的?件；Shell命令實(shí)現(xiàn)：hadoopfs-rm/Test/test1/text1.txt（10）刪除HDFS中指定的?錄，由?戶指定?錄中如果存在?件時是否刪除?錄；Shell命令實(shí)現(xiàn)：hadoopfs-rm-R/Test/test1（11）在HDFS中，將?件從源路徑移動到?的路徑。hadoopfs-mvtJAVA實(shí)現(xiàn)：importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.*;importorg.apache.hadoop.fs.FileSystem;importjava.io.*;publicclassMoveFile{/***移動?件*/publicstaticbooleanmv(Configurationconf,StringremoteFilePath,StringremoteToFilePath){try(FileSystemfs=FileSystem.get(conf)){PathsrcPath=newPath(remoteFilePath);PathdstPath=newPath(remoteToFilePath);returnfs.rename(srcPath,dstPath);}catch(IOExceptione){e.printStackTrace();returnfalse;}}/***主函數(shù)*/publicstaticvoidmain(String[]args){Configurationconf=newConfiguration();conf.set("fs.defaultFS","hdfs://localhost:9000");StringremoteFilePath="hdfs:///user/tiny/text.txt";//源?件HDFS路徑StringremoteToFilePath="hdfs:///user/tiny/input";//?的HDFS路徑try{if(MoveFile.mv(conf,remoteFilePath,remoteToFilePath)){System.out.println("將?件"+remoteFilePath+"移動到"+remoteToFilePath);}else{System.out.println("操作失敗(源?件不存在或移動失敗)");}}catch(Exceptione){e.printStackTrace();}}}2.編程實(shí)現(xiàn)?個類“MyFSDataInputStream”，該類繼承“org.apache.hadoop.fs.FSDataInputStream”，要求如下：實(shí)現(xiàn)按?讀取HDFS中指定?件的?法“readLine()”，如果讀到?件末尾，則返回空，否則返回?件??的?本。JAVA實(shí)現(xiàn)：importjava.io.BufferedReader;importjava.io.IOException;importjava.io.InputStream;importjava.io.InputStreamReader;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.*;publicclassMyFSDataInputStreamextendsFSDataInputStream{/***@paramargs*/public