使用實例：

1.項目結構（引入包hadoop-0.20.2-core.jar和commons-logging.jar）

2.代碼

HdfsCommon.java

?

public class HdfsCommon {private Configuration conf;private FileSystem fs;public HdfsCommon() throws IOException{conf=new Configuration();fs=FileSystem.get(conf);}/*** 上傳文件，* @param localFile 本地路徑* @param hdfsPath 格式為hdfs://ip:port/destination* @throws IOException*/public void upFile(String localFile,String hdfsPath) throws IOException{InputStream in=new BufferedInputStream(new FileInputStream(localFile));OutputStream out=fs.create(new Path(hdfsPath));IOUtils.copyBytes(in, out, conf);}/*** 附加文件* @param localFile* @param hdfsPath* @throws IOException*/public void appendFile(String localFile,String hdfsPath) throws IOException{InputStream in=new FileInputStream(localFile);OutputStream out=fs.append(new Path(hdfsPath));IOUtils.copyBytes(in, out, conf);}/*** 下載文件* @param hdfsPath* @param localPath* @throws IOException*/public void downFile(String hdfsPath, String localPath) throws IOException{InputStream in=fs.open(new Path(hdfsPath));OutputStream out=new FileOutputStream(localPath);IOUtils.copyBytes(in, out, conf);}/*** 刪除文件或目錄* @param hdfsPath* @throws IOException*/public void delFile(String hdfsPath) throws IOException{fs.delete(new Path(hdfsPath), true);} }

core-site.xml

?

?

<?xml version="1.0"?> <?xml-stylesheet type="text/xsl" href="configuration.xsl"?> <configuration> <!-- global properties --><property><name>hadoop.tmp.dir</name><value>/home/whuqin/tmp</value></property><!-- file system properties --><property><name>fs.default.name</name><value>hdfs://localhost:9000</value></property> </configuration>

測試代碼Test.java

?

?

public class Test {public static void main(String[] args) throws IOException {HdfsCommon hdfs=new HdfsCommon(); // hdfs.upFile("/home/whuqin/file01", "hdfs://localhost:9000/user/whuqin/input/file01copy1"); // hdfs.downFile("hdfs://localhost:9000/user/whuqin/input/file01copy", "/home/whuqin/fileCopy"); // hdfs.appendFile("/home/whuqin/file01", "hdfs://localhost:9000/user/whuqin/input/file01copy");hdfs.delFile("hdfs://localhost:9000/user/whuqin/input/file01copy1");} }

3.直接點擊項目，右鍵運行即可(在eclipse下)

?

4. 在終端下，使用hadoop命令行測試結果。

讀文件流程：

1.client調用FileSystem.open()方法：

FileSystem通過RPC與namenode通信，namenode返回該文件的部分或全部block列表（含有block拷貝的datanode地址）。

選取距離客戶端最近的datanode建立連接，讀取block，返回FSDataInputStream。

2.client調用輸入流的read()方法：

當讀到block結尾時，FSDataInputStream關閉與當前datanode的連接，并為讀取下一個block尋找最近datanode。

讀取完一個block都會進行checksum驗證，如果讀取datanode時出現錯誤，客戶端會通知Namenode，然后再從下一個擁有該block拷貝的datanode繼續(xù)讀。

如果block列表讀完后，文件還未結束，FileSystem會繼續(xù)從namenode獲取下一批block列表。

這些操作對client透明，client感覺到的是連續(xù)的流。

3.關閉FSDataInputStream

寫文件流程：

1.client調用FileSystem的create()方法：

FileSystem向namenode發(fā)出請求，在namenode的namespace里面創(chuàng)建一?新文件，但是并不關聯任何塊。

Namenode檢查文件是否已存在、操作權限。如果檢查通過，namenode記錄新文件信息，并在某一個datanode上創(chuàng)建數據塊。

返回FSDataOutputStream，將client引導至該數據塊執(zhí)行寫入操作。

2.client調用輸出流的write()方法：HDFS默認將每個數據塊放置3份。FSDataOutputStream將數據首先寫到第一節(jié)點，第一節(jié)點將數據包傳送并寫入第二節(jié)點，第二節(jié)點=》第三節(jié)點。

3.client調用流的close()方法：flush緩沖區(qū)的數據包，block完成復制份數后，namenode返回成功消息。

補：1.客戶端可以獨立于Hadoop的集群。2.hadoop的底層傳輸協(xié)議為RPC。3.寫數據備份放置策略：如果客戶端在集群上，第一份在客戶端，否則隨機；第二份與第一份不在同一機架；第三份與第二份在同一機架，不同節(jié)點；其余放在隨機節(jié)點，但避免一個機架多份備份。

?

轉載于:https://www.cnblogs.com/whuqin/archive/2011/08/29/4982058.html

總結

以上是生活随笔為你收集整理的HDFS读写文件实例与解析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。