hdfsjava寫文件

① HDFS文件

Hadoop支持的文件系統由很多（見下圖），HDFS只是其中一種實現。java抽象類 org.apache.hadoop.fs.FileSystem 定義了Hadoop中一個文件系統的客戶端介面，並且該抽象類有幾個具體實現。Hadoop一般使用URI（下圖）方案來選取合適的文件系統實例進行交互。

特別的，HDFS文件系統的操作可以使用 FsSystem shell 、客戶端（http rest api、Java api、C api等）。

FsSystem shell 的用法基本同本地shell類似，命令可參考 FsSystem shell

Hadoop是用Java寫的，通過Java Api（ FileSystem 類）可以調用大部分Hadoop文件系統的交互操作。更詳細的介紹可參考 hadoop Filesystem 。

非Java開發的應用可以使用由WebHDFS協議提供的HTTP REST API，但是HTTP比原生的Java客戶端要慢，所以不到萬不得已盡量不要使用HTTP傳輸特大數據。通過HTTP來訪問HDFS有兩種方法：

兩種如圖

在第一種情況中，namenode和datanode內嵌的web服務作為WebHDFS的端節點運行（是否啟用WebHDFS可通過dfs.webhdfs.enabled設置，默認為true）。文件元數據在namenode上，文件讀寫操作首先被發往namenode，有namenode發送一個HTTP重定向至某個客戶端，指示以流的方式傳輸文件數據的目的或源datanode。

第二種方法依靠一個或多個獨立代理伺服器通過HTTP訪問HDFS。所有集群的網路通信都需要通過代理，因此客戶端從來不直接訪問namenode或datanode。使用代理後可以使用更嚴格的防火牆策略和帶寬策略。

HttpFs代理提供和WebHDFS相同的HTTP介面，這樣客戶端能夠通過webhdfs URI訪問介面。HttpFS代理啟動獨立於namenode和datanode的守護進程，使用httpfs.sh 腳本，默認在一個不同的埠上監聽（14000）。

下圖描述了

讀文件時客戶端與 HDFS 中的 namenode， datanode 之間的數據流動。

對上圖的解釋如下：

在讀取過程中， 如果 FSDataInputStream 在和一個 datanode 進行交流時出現了一個錯誤，他就去試一試下一個最接近的塊，他當然也會記住剛才發生錯誤的 datanode 以至於之後不會再在這個 datanode 上進行沒必要的嘗試。 DFSInputStream 也會在 datanode 上傳輸出的數據上核查檢查數(checknums).如果損壞的塊被發現了， DFSInputStream 就試圖從另一個擁有備份的 datanode 中去讀取備份塊中的數據。

在這個設計中一個重要的方面就是客戶端直接從 datanode 上檢索數據，並通過 namenode 指導來得到每一個塊的最佳 datanode。這種設計允許 HDFS 擴展大量的並發客戶端，因為數據傳輸只是集群上的所有 datanode 展開的。期間，namenode 僅僅只需要服務於獲取塊位置的請求（塊位置信息是存放在內存中，所以效率很高）。如果不這樣設計，隨著客戶端數據量的增長，數據服務就會很快成為一個瓶頸。

我們知道，相對於客戶端(之後就是 maprece task 了)，塊的位置有以下可能性：

我們認為他們對於客戶端的帶寬遞減，距離遞增（括弧中表示距離）。示意圖如下：

如果集群中的機器都在同一個機架上，我們無需其他配置，若集群比較復雜，由於hadoop無法自動發現網路拓撲，所以需要額外配置網路拓撲。

基本讀取程序，將文件內容輸出到console

FileSystemCat

隨機讀取

展開原碼

下圖描述了寫文件時客戶端與 HDFS 中的 namenode， datanode 之間的數據流動。

對上圖的解釋如下：

如果在任何一個 datanode 在寫入數據的時候失敗了，接下來所做的一切對客戶端都是透明的：首先， pipeline 被關閉，在確認隊列中的剩下的包會被添加進數據隊列的起始位置上，以至於在失敗的節點下游的任 何節點都不會丟失任何的包。然後與 namenode 聯系後，當前在一個好的 datanode 會聯系 namenode， 給失敗節點上還未寫完的塊生成一個新的標識ID， 以至於如果這個失敗的 datanode 不久後恢復了，這個不完整的塊將會被刪除。失敗節點會從 pipeline 中移除，然後剩下兩個好的 datanode 會組成一個的新的 pipeline ，剩下的 這些塊的包(也就是剛才放在數據隊列隊首的包)會繼續寫進 pipeline 中好的 datanode 中。最後，namenode 注意到塊備份數小於規定的備份數，他就安排在另一個節點上創建完成備份，直接從已有的塊中復制就可以。然後一直到滿足了備份數( dfs.replication )。如果有多個節點的寫入失敗了，如果滿足了最小備份數的設置( dfs.namenode.repliction.min ),寫入也將會成功，然後剩下的備份會被集群非同步的執行備份，直到滿足了備份數( dfs.replication )。

創建目錄

文件壓縮有兩大好處：

Hadoop 對於壓縮格式的是自動識別。如果我們壓縮的文件有相應壓縮格式的擴展名（比如 lzo，gz，bzip2 等）。Hadoop 會根據壓縮格式的擴展名自動選擇相對應的解碼器來解壓數據，此過程完全是 Hadoop 自動處理，我們只需要確保輸入的壓縮文件有擴展名。

Hadoop中有多種壓縮格式、演算法和工具，下圖列出了常用的壓縮方法。

表中的「是否可切分」表示對應的壓縮演算法是否支持切分，也就是說是否可以搜索數據流的任意位置並進一步往下讀取數據，可切分的壓縮格式尤其適合MapRece。

所有的壓縮演算法都需要權衡空間/時間：壓縮和解壓縮速度更快，其代價通常是只能節省少量的空間。不同的壓縮工具有不同的特性：

更詳細的比較如下

1.壓縮性能比較

2.優缺點

另外使用hadoop原生（native）類庫比其他java實現有更快的壓縮和解壓縮速度。特徵比較如下：

使用容器文件格式結合壓縮演算法也能更好的提高效率。順序文件、Arvo文件、ORCFiles、Parqurt文件同時支持壓縮和切分。

壓縮舉例（Java）

壓縮

解壓縮

六、文件序列化

序列化是指將結構化數據轉換為位元組流以便在網路上傳輸或寫到磁碟進行永久存儲。反序列化獅子將位元組流轉換回結構化對象的逆過程。

序列化用於分布式數據處理的兩大領域：進程間通信和永久存儲。

對序列化的要求時是格式緊湊（高效使用存儲空間）、快速（讀寫效率高）、可擴展（可以透明地讀取老格式數據）且可以互操作（可以使用不同的語言讀寫數據）。

Hadoop使用的是自己的序列化格式 Writable ，它絕對緊湊、速度快，但不太容易用java以外的語言進行擴展或使用。

當然，用戶也可以使用其他序列化框架或者自定義序列化方式，如 Avro 框架。

Hadoop內部還使用了 Apache Thrift 和 Protocal Buffers 來實現RPC和數據交換。

② 關於用java寫程序把本地文件上傳到HDFS中的問題

將這FileSystem hdfs = FileSystem.get(config);
改成FileSystem hdfs = FileSystem.get(URI.create("hdfs://master:9000"),config)
上面那句取得的是本地文件系統對象，改成下面這個才是取得hdfs文件系統對象，當你要操作本地文件對象的時候就要用上面那句取得本地文件對象，我在2.7.4剛開始也是跟你一樣的錯誤，改為下面的就可以了

③ 如何使用Java API讀寫HDFS

Java API讀寫HDFS

public class FSOptr {

/**
* @param args
*/
public static void main(String[] args) throws Exception {
// TODO Auto-generated method stub
Configuration conf = new Configuration();
makeDir(conf);
rename(conf);
delete(conf);

}

// 創建文件目錄
private static void makeDir(Configuration conf) throws Exception {
FileSystem fs = FileSystem.get(conf);
Path dir = new Path("/user/hadoop/data/20140318");
boolean result = fs.mkdirs(dir);// 創建文件夾
System.out.println("make dir :" + result);

// 創建文件，並寫入內容
Path dst = new Path("/user/hadoop/data/20140318/tmp");
byte[] buff = "hello,hadoop!".getBytes();
FSDataOutputStream outputStream = fs.create(dst);
outputStream.write(buff, 0, buff.length);
outputStream.close();
FileStatus files[] = fs.listStatus(dst);
for (FileStatus file : files) {
System.out.println(file.getPath());
}
fs.close();
}

// 重命名文件
private static void rename(Configuration conf) throws Exception {

FileSystem fs = FileSystem.get(conf);
Path oldName = new Path("/user/hadoop/data/20140318/1.txt");
Path newName = new Path("/user/hadoop/data/20140318/2.txt");
fs.rename(oldName, newName);

FileStatus files[] = fs.listStatus(new Path(
"/user/hadoop/data/20140318"));
for (FileStatus file : files) {
System.out.println(file.getPath());
}
fs.close();
}

// 刪除文件
@SuppressWarnings("deprecation")
private static void delete(Configuration conf) throws Exception {
FileSystem fs = FileSystem.get(conf);
Path path = new Path("/user/hadoop/data/20140318");
if (fs.isDirectory(path)) {
FileStatus files[] = fs.listStatus(path);
for (FileStatus file : files) {
fs.delete(file.getPath());
}
} else {
fs.delete(path);
}

// 或者
fs.delete(path, true);

fs.close();
}

/**
* 下載,將hdfs文件下載到本地磁碟
*
* @param localSrc1
* 本地的文件地址，即文件的路徑
* @param hdfsSrc1
* 存放在hdfs的文件地址
*/
public boolean sendFromHdfs(String hdfsSrc1, String localSrc1) {

Configuration conf = new Configuration();
FileSystem fs = null;
try {
fs = FileSystem.get(URI.create(hdfsSrc1), conf);
Path hdfs_path = new Path(hdfsSrc1);
Path local_path = new Path(localSrc1);

fs.ToLocalFile(hdfs_path, local_path);

return true;
} catch (IOException e) {
e.printStackTrace();
}
return false;
}

/**
* 上傳，將本地文件到hdfs系統中
*
* @param localSrc
* 本地的文件地址，即文件的路徑
* @param hdfsSrc
* 存放在hdfs的文件地址
*/
public boolean sendToHdfs1(String localSrc, String hdfsSrc) {
InputStream in;
try {
in = new BufferedInputStream(new FileInputStream(localSrc));
Configuration conf = new Configuration();// 得到配置對象
FileSystem fs; // 文件系統
try {
fs = FileSystem.get(URI.create(hdfsSrc), conf);
// 輸出流，創建一個輸出流
OutputStream out = fs.create(new Path(hdfsSrc),
new Progressable() {
// 重寫progress方法
public void progress() {
// System.out.println("上傳完一個設定緩存區大小容量的文件！");
}
});
// 連接兩個流，形成通道，使輸入流向輸出流傳輸數據,
IOUtils.Bytes(in, out, 10240, true); // in為輸入流對象，out為輸出流對象，4096為緩沖區大小，true為上傳後關閉流
return true;
} catch (IOException e) {
e.printStackTrace();
}

} catch (FileNotFoundException e) {
e.printStackTrace();
}
return false;
}

/**
* 移動
*
* @param old_st原來存放的路徑
* @param new_st移動到的路徑
*/
public boolean moveFileName(String old_st, String new_st) {

try {

// 下載到伺服器本地
boolean down_flag = sendFromHdfs(old_st, "/home/hadoop/文檔/temp");
Configuration conf = new Configuration();
FileSystem fs = null;

// 刪除源文件
try {
fs = FileSystem.get(URI.create(old_st), conf);
Path hdfs_path = new Path(old_st);
fs.delete(hdfs_path);
} catch (IOException e) {
e.printStackTrace();
}

// 從伺服器本地傳到新路徑
new_st = new_st + old_st.substring(old_st.lastIndexOf("/"));
boolean uplod_flag = sendToHdfs1("/home/hadoop/文檔/temp", new_st);

if (down_flag && uplod_flag) {
return true;
}
} catch (Exception e) {
e.printStackTrace();
}
return false;
}

// 本地文件到hdfs
private static void CopyFromLocalFile(Configuration conf) throws Exception {
FileSystem fs = FileSystem.get(conf);
Path src = new Path("/home/hadoop/word.txt");
Path dst = new Path("/user/hadoop/data/");
fs.FromLocalFile(src, dst);
fs.close();
}

// 獲取給定目錄下的所有子目錄以及子文件
private static void getAllChildFile(Configuration conf) throws Exception {
FileSystem fs = FileSystem.get(conf);
Path path = new Path("/user/hadoop");
getFile(path, fs);
}

private static void getFile(Path path, FileSystem fs)throws Exception {
FileStatus[] fileStatus = fs.listStatus(path);
for (int i = 0; i < fileStatus.length; i++) {
if (fileStatus[i].isDir()) {
Path p = new Path(fileStatus[i].getPath().toString());
getFile(p, fs);
} else {
System.out.println(fileStatus[i].getPath().toString());
}
}
}

//判斷文件是否存在
private static boolean isExist(Configuration conf,String path)throws Exception{
FileSystem fileSystem = FileSystem.get(conf);
return fileSystem.exists(new Path(path));
}

//獲取hdfs集群所有主機結點數據
private static void getAllClusterNodeInfo(Configuration conf)throws Exception{
FileSystem fs = FileSystem.get(conf);
DistributedFileSystem hdfs = (DistributedFileSystem)fs;
DatanodeInfo[] dataNodeStats = hdfs.getDataNodeStats();
String[] names = new String[dataNodeStats.length];
System.out.println("list of all the nodes in HDFS cluster:"); //print info

for(int i=0; i < dataNodeStats.length; i++){
names[i] = dataNodeStats[i].getHostName();
System.out.println(names[i]); //print info

}
}

//get the locations of a file in HDFS
private static void getFileLocation(Configuration conf)throws Exception{
FileSystem fs = FileSystem.get(conf);
Path f = new Path("/user/cluster/dfs.txt");
FileStatus filestatus = fs.getFileStatus(f);
BlockLocation[] blkLocations = fs.getFileBlockLocations(filestatus,0,filestatus.getLen());
int blkCount = blkLocations.length;
for(int i=0; i < blkCount; i++){
String[] hosts = blkLocations[i].getHosts();
//Do sth with the block hosts

System.out.println(hosts);
}
}

//get HDFS file last modification time
private static void getModificationTime(Configuration conf)throws Exception{
FileSystem fs = FileSystem.get(conf);
Path f = new Path("/user/cluster/dfs.txt");
FileStatus filestatus = fs.getFileStatus(f);

long modificationTime = filestatus.getModificationTime(); // measured in milliseconds since the epoch

Date d = new Date(modificationTime);
System.out.println(d);
}

}

④ java怎麼連接hdfs文件系統，需要哪些包

apache的Hadoop項目提供一類api可以通過java工程操作hdfs中的文件，包括：文件打開，讀寫，刪除等、目錄的創建，刪除，讀取目錄中所有文件等。
1、到http://hadoop.apache.org/releases.html下載Hadoop，解壓後把所有jar加入項目的lib里
2、程序處理步驟： 1）得到Configuration對象，2）得到FileSystem對象，3）進行文件操作，簡單示例如下：
/**
*
*/
package org.jrs.wlh;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

/**
* @PutMeger.java
* java操作hdfs 往 hdfs中上傳數據
* @version $Revision$</br>
* update: $Date$
*/
public class PutMeger {

public static void main(String[] args) throws IOException {

String[] str = new String[]{"E:\\hadoop\\UploadFileClient.java","hdfs://master:9000/user/hadoop/inccnt.java"};
Configuration conf = new Configuration();
FileSystem fileS= FileSystem.get(conf);
FileSystem localFile = FileSystem.getLocal(conf); //得到一個本地的FileSystem對象

Path input = new Path(str[0]); //設定文件輸入保存路徑
Path out = new Path(str[1]); //文件到hdfs輸出路徑

try{
FileStatus[] inputFile = localFile.listStatus(input); //listStatus得到輸入文件路徑的文件列表
FSDataOutputStream outStream = fileS.create(out); //創建輸出流
for (int i = 0; i < inputFile.length; i++) {
System.out.println(inputFile[i].getPath().getName());
FSDataInputStream in = localFile.open(inputFile[i].getPath());

byte buffer[] = new byte[1024];
int bytesRead = 0;
while((bytesRead = in.read(buffer))>0){ //按照位元組讀取數據
System.out.println(buffer);
outStream.write(buffer,0,bytesRead);
}

in.close();
}

}catch(Exception e){
e.printStackTrace();
}
}

}

⑤ 編寫一個JAVA類方法，通過該方法可以獲取出存儲在HDFS集群中根目錄的所有文件

public void listMyFile() throws Exception {

//獲取FileSystem
//"hdfs"為偽造用戶，使用hdfs用戶進行訪問
FileSystem fileSystem = FileSystem.get(new URI("hdfs://192.168.83.141:8020"), new Configuration(), "hdfs");
//獲取指定目標目錄下的所有文件信息
RemoteIterator<LocatedFileStatus> iterator =
fileSystem.listFiles(new Path("/"), true);
//遍歷迭代器
while (iterator.hasNext()) {
//獲取每個文件詳細信息
LocatedFileStatus fileStatus = iterator.next();
//獲取每個文件的存儲路徑
System.out.println("路徑：" + fileStatus.getPath() +
"---" + fileStatus.getPath().getName());
//獲取文件的block存儲信息
BlockLocation[] blockLocations = fileStatus.getBlockLocations();
//列印每個文件的block數
System.out.println("block數量：" + blockLocations.length);
//列印每一個block副本的存儲位置
for (BlockLocation blockLocation : blockLocations) {
String[] hosts = blockLocation.getHosts();
for (String host : hosts) {
System.out.println("主機：" + host);
}
}
}
}

⑥ 如何使用Java API讀寫HDFS

package com.wyc.hadoop.fs;

import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.util.Date;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.DistributedFileSystem;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;

public class FSOptr {

/**
* @param args
*/
public static void main(String[] args) throws Exception {
// TODO Auto-generated method stub
Configuration conf = new Configuration();
makeDir(conf);
rename(conf);
delete(conf);

}

// 創建文件目錄
private static void makeDir(Configuration conf) throws Exception {
FileSystem fs = FileSystem.get(conf);
Path dir = new Path("/user/hadoop/data/20140318");
boolean result = fs.mkdirs(dir);// 創建文件夾
System.out.println("make dir :" + result);

// 創建文件，並寫入內容
Path dst = new Path("/user/hadoop/data/20140318/tmp");
byte[] buff = "hello,hadoop!".getBytes();
FSDataOutputStream outputStream = fs.create(dst);
outputStream.write(buff, 0, buff.length);
outputStream.close();
FileStatus files[] = fs.listStatus(dst);
for (FileStatus file : files) {
System.out.println(file.getPath());
}
fs.close();
}

// 重命名文件
private static void rename(Configuration conf) throws Exception {

FileSystem fs = FileSystem.get(conf);
Path oldName = new Path("/user/hadoop/data/20140318/1.txt");
Path newName = new Path("/user/hadoop/data/20140318/2.txt");
fs.rename(oldName, newName);

FileStatus files[] = fs.listStatus(new Path(
"/user/hadoop/data/20140318"));
for (FileStatus file : files) {
System.out.println(file.getPath());
}
fs.close();
}

// 刪除文件
@SuppressWarnings("deprecation")
private static void delete(Configuration conf) throws Exception {
FileSystem fs = FileSystem.get(conf);
Path path = new Path("/user/hadoop/data/20140318");
if (fs.isDirectory(path)) {
FileStatus files[] = fs.listStatus(path);
for (FileStatus file : files) {
fs.delete(file.getPath());
}
} else {
fs.delete(path);
}

// 或者
fs.delete(path, true);

fs.close();
}

/**
* 下載,將hdfs文件下載到本地磁碟
*
* @param localSrc1
* 本地的文件地址，即文件的路徑
* @param hdfsSrc1
* 存放在hdfs的文件地址
*/
public boolean sendFromHdfs(String hdfsSrc1, String localSrc1) {

Configuration conf = new Configuration();
FileSystem fs = null;
try {
fs = FileSystem.get(URI.create(hdfsSrc1), conf);
Path hdfs_path = new Path(hdfsSrc1);
Path local_path = new Path(localSrc1);

fs.ToLocalFile(hdfs_path, local_path);

return true;
} catch (IOException e) {
e.printStackTrace();
}
return false;
}

/**
* 上傳，將本地文件到hdfs系統中
*
* @param localSrc
* 本地的文件地址，即文件的路徑
* @param hdfsSrc
* 存放在hdfs的文件地址
*/
public boolean sendToHdfs1(String localSrc, String hdfsSrc) {
InputStream in;
try {
in = new BufferedInputStream(new FileInputStream(localSrc));
Configuration conf = new Configuration();// 得到配置對象
FileSystem fs; // 文件系統
try {
fs = FileSystem.get(URI.create(hdfsSrc), conf);
// 輸出流，創建一個輸出流
OutputStream out = fs.create(new Path(hdfsSrc),
new Progressable() {
// 重寫progress方法
public void progress() {
// System.out.println("上傳完一個設定緩存區大小容量的文件！");
}
});
// 連接兩個流，形成通道，使輸入流向輸出流傳輸數據,
IOUtils.Bytes(in, out, 10240, true); // in為輸入流對象，out為輸出流對象，4096為緩沖區大小，true為上傳後關閉流
return true;
} catch (IOException e) {
e.printStackTrace();
}

} catch (FileNotFoundException e) {
e.printStackTrace();
}
return false;
}

/**
* 移動
*
* @param old_st原來存放的路徑
* @param new_st移動到的路徑
*/
public boolean moveFileName(String old_st, String new_st) {

try {

// 下載到伺服器本地
boolean down_flag = sendFromHdfs(old_st, "/home/hadoop/文檔/temp");
Configuration conf = new Configuration();
FileSystem fs = null;

// 刪除源文件
try {
fs = FileSystem.get(URI.create(old_st), conf);
Path hdfs_path = new Path(old_st);
fs.delete(hdfs_path);
} catch (IOException e) {
e.printStackTrace();
}

// 從伺服器本地傳到新路徑
new_st = new_st + old_st.substring(old_st.lastIndexOf("/"));
boolean uplod_flag = sendToHdfs1("/home/hadoop/文檔/temp", new_st);

if (down_flag && uplod_flag) {
return true;
}
} catch (Exception e) {
e.printStackTrace();
}
return false;
}

// 本地文件到hdfs
private static void CopyFromLocalFile(Configuration conf) throws Exception {
FileSystem fs = FileSystem.get(conf);
Path src = new Path("/home/hadoop/word.txt");
Path dst = new Path("/user/hadoop/data/");
fs.FromLocalFile(src, dst);
fs.close();
}

// 獲取給定目錄下的所有子目錄以及子文件
private static void getAllChildFile(Configuration conf) throws Exception {
FileSystem fs = FileSystem.get(conf);
Path path = new Path("/user/hadoop");
getFile(path, fs);
}

private static void getFile(Path path, FileSystem fs)throws Exception {
FileStatus[] fileStatus = fs.listStatus(path);
for (int i = 0; i < fileStatus.length; i++) {
if (fileStatus[i].isDir()) {
Path p = new Path(fileStatus[i].getPath().toString());
getFile(p, fs);
} else {
System.out.println(fileStatus[i].getPath().toString());
}
}
}

//判斷文件是否存在
private static boolean isExist(Configuration conf,String path)throws Exception{
FileSystem fileSystem = FileSystem.get(conf);
return fileSystem.exists(new Path(path));
}

//獲取hdfs集群所有主機結點數據
private static void getAllClusterNodeInfo(Configuration conf)throws Exception{
FileSystem fs = FileSystem.get(conf);
DistributedFileSystem hdfs = (DistributedFileSystem)fs;
DatanodeInfo[] dataNodeStats = hdfs.getDataNodeStats();
String[] names = new String[dataNodeStats.length];
System.out.println("list of all the nodes in HDFS cluster:"); //print info

for(int i=0; i < dataNodeStats.length; i++){
names[i] = dataNodeStats[i].getHostName();
System.out.println(names[i]); //print info

}
}

//get the locations of a file in HDFS
private static void getFileLocation(Configuration conf)throws Exception{
FileSystem fs = FileSystem.get(conf);
Path f = new Path("/user/cluster/dfs.txt");
FileStatus filestatus = fs.getFileStatus(f);
BlockLocation[] blkLocations = fs.getFileBlockLocations(filestatus,0,filestatus.getLen());
int blkCount = blkLocations.length;
for(int i=0; i < blkCount; i++){
String[] hosts = blkLocations[i].getHosts();
//Do sth with the block hosts

System.out.println(hosts);
}
}

//get HDFS file last modification time
private static void getModificationTime(Configuration conf)throws Exception{
FileSystem fs = FileSystem.get(conf);
Path f = new Path("/user/cluster/dfs.txt");
FileStatus filestatus = fs.getFileStatus(f);

long modificationTime = filestatus.getModificationTime(); // measured in milliseconds since the epoch

Date d = new Date(modificationTime);
System.out.println(d);
}

}