在項目開發(fā)中风响，有時我們需要通過HDFS的api來對文件進行操作袱衷，比如將數(shù)據(jù)上傳到HDFS或者從HDFS獲取數(shù)據(jù)等敬察。本篇來介紹一下HDFS中API的具體使用薪捍。直接上代碼：
package com.lzb.hdfs.fs;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import java.io.IOException;
public class HDFSHelper {
private FileSystem fs;
public HDFSHelper() {
fs = getFileSystem();
}
/**
* Configuration是配置對象，conf可以理解為包含了所有配置信息的一個集合抽兆，可以認為是Map,
* 在初始化的時候底層會加載一堆配置文件 core-site.xml;hdfs-site.xml;mapred-site.xml;yarn-site.xml
* 如果需要項目代碼自動加載配置文件中的信息识补，那么就必須把配置文件改成-default.xml或者-site.xml的名稱，
* 而且必須放置在src下,如果不叫這個名郊丛，或者不在src下李请，也需要加載這些配置文件中的參數(shù)瞧筛，必須使用conf對象提供的方法手動加載.
* 依次加載的參數(shù)信息的順序是：
* 1.加載core/hdfs/mapred/yarn-default.xml
* 2.加載通過conf.addResource()加載的配置文件
* 3.加載conf.set(name,value)
*/
private Configuration getConfiguration(){
Configuration conf = new Configuration();
//conf.addResource("xxx");
//conf.set("xxx","xxx");
//Configuration.addDefaultResource("core-site.xml");
//Configuration.addDefaultResource("hdfs-site.xml");
//conf.set("fs.default.name","hdfs://probd01:8020");
//HA模式的配置
conf.set("fs.defaultFS", "hdfs://probd");
conf.set("dfs.nameservices", "probd");
conf.set("dfs.ha.namenodes.probd", "nn1,nn2");
conf.set("dfs.namenode.rpc-address.probd.nn1", "probd01:8020");
conf.set("dfs.namenode.rpc-address.probd.nn2", "probd02:8020");
conf.set("dfs.client.failover.proxy.provider.probd", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");
//防止報錯：no FileSystem for scheme: hdfs...
conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
return conf;
}
/**
* 獲取文件系統(tǒng)
* 本地文件系統(tǒng)為LocalFileSystem，URL形式: file:///c:myProgram
* HDFS文件系統(tǒng)為DistributedFileSystem导盅，URL形式: fs.defaultFS=hdfs://hadoop01:9000
*/
public FileSystem getFileSystem(){
Configuration conf = getConfiguration();
FileSystem fs = null;
try {
fs = FileSystem.get(conf);
} catch (IOException e) {
e.printStackTrace();
}
System.out.println(fs);
return fs;
}
/**
* 上傳本地文件到HDFS较幌，底層就是采用流的方式
* @param localPath 本地文件路徑
* @param remotePath HDFS文件路徑
* @return 是否上傳成功
*/
public boolean copyFromLocal(String localPath,String remotePath){
if(fs == null) return false;
try {
fs.copyFromLocalFile(new Path(localPath),new Path(remotePath));
} catch (IOException e) {
e.printStackTrace();
}
return true;
}
/**
* 從HDFS下載文件，底層就是采用流的方式
* @param remotePath HDFS文件路徑
* @param localPath 本地路徑
* @return 是否下載成功
*/
public boolean copyToLocal(String remotePath,String localPath){
if(fs == null) return false;
try {
fs.copyToLocalFile(new Path(remotePath),new Path(localPath));
} catch (IOException e) {
e.printStackTrace();
return false;
}
return true;
}
/**
* 獲取目錄下的文件
* @param remotePath HDFS文件路徑
* @param recursive 是否級聯(lián)（該文件夾下面如果還有子文件 要不要看,注意沒有 子文件夾!!）
*/
public void listFiles(String remotePath,boolean recursive){
if(fs == null) return;
try {
RemoteIterator<LocatedFileStatus> iterator = fs.listFiles(new Path(remotePath), recursive);
while (iterator.hasNext()){
LocatedFileStatus fileStatus = iterator.next();
//文件的存儲路徑白翻，以hdfs://開頭的全路徑 ==> hdfs://hadoop01:9000/a/gg.txt
System.out.println( "file path === " + fileStatus.getPath());
//文件名
System.out.println("file name === " + fileStatus.getPath().getName());
//文件長度
System.out.println("file size === "+fileStatus.getLen());
//文件所有者
System.out.println("file owner === "+fileStatus.getOwner());
//分組信息
System.out.println("file group === " + fileStatus.getGroup());
//文件權(quán)限信息
System.out.println("file permission === " + fileStatus.getPermission());
//文件副本數(shù)
System.out.println("file blocks === " + fileStatus.getReplication());
//塊大小
System.out.println("file block size === " + fileStatus.getBlockSize());
//塊位置相關信息
BlockLocation[] blockLocations = fileStatus.getBlockLocations();
//塊的數(shù)量
System.out.println("file block nums === " + blockLocations.length);
for (BlockLocation bl : blockLocations) {
String[] hosts = bl.getHosts();
for (String host: hosts) {
System.out.println("block host === " + host);
}
//塊的一個邏輯路徑
bl.getTopologyPaths();
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 獲取目錄下的文件
* 此方法與listFiles不同,不支持傳true或false,即不能級聯(lián)乍炉，如果想實現(xiàn)級聯(lián)就采用遞歸的方式
* @param remotePath HDFS文件路徑
*/
public void listStatus(String remotePath){
if(fs == null) return;
try {
FileStatus[] listStatus = fs.listStatus(new Path(remotePath));
for (FileStatus fss : listStatus) {
//判斷是不是文件夾
boolean directory = fss.isDirectory();
//判斷是不是文件
boolean file = fss.isFile();
String name = fss.getPath().getName();
if(file) {
System.out.println(name+":文件");
}else {
System.out.println(name+":文件夾");
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 刪除空文件夾或空文件
* @param path
*/
public void deleteEmptyDirAndFile(Path path){
if(fs == null) return;
try {
FileStatus[] listStatus = fs.listStatus(path);
if(listStatus.length == 0){
//刪除空文件夾
fs.delete(path,true);
return;
}
RemoteIterator<LocatedFileStatus> iterator = fs.listLocatedStatus(path);
while (iterator.hasNext()) {
LocatedFileStatus next = iterator.next();
Path currentPath = next.getPath();
Path parentPath = next.getPath().getParent();
if (next.isDirectory()) {
// 如果是空文件夾
if (fs.listStatus(currentPath).length == 0) {
// 刪除掉
fs.delete(currentPath, true);
} else {
// 不是空文件夾，那么則繼續(xù)遍歷
if (fs.exists(currentPath)) {
deleteEmptyDirAndFile(currentPath);
}
}
} else {
// 獲取文件的長度
long fileLength = next.getLen();
// 當文件是空文件時滤馍， 刪除
if (fileLength == 0) {
fs.delete(currentPath, true);
}
}
// 當空文件夾或者空文件刪除時岛琼，有可能導致父文件夾為空文件夾，
// 所以每次刪除一個空文件或者空文件的時候都需要判斷一下巢株，如果真是如此槐瑞，那么就需要把該文件夾也刪除掉
int length = fs.listStatus(parentPath).length;
if (length == 0) {
fs.delete(parentPath, true);
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 創(chuàng)建文件夾
* @param remotePath HDFS文件路徑
* @return 是否創(chuàng)建成功
*/
public boolean mkdir(String remotePath){
if(fs == null) return false;
boolean success = false;
try {
success = fs.mkdirs(new Path(remotePath));
} catch (IOException e) {
e.printStackTrace();
}
return success;
}
/**
* 寫入文件
* @param remotePath HDFS文件路徑
* @param content 內(nèi)容
* @return 是否寫入成功
*/
public boolean writeToFile(String remotePath,String content){
if(fs == null) return false;
try {
FSDataOutputStream out = fs.create(new Path(remotePath));
out.writeUTF(content);
out.close();
} catch (IOException e) {
e.printStackTrace();
return false;
}
return true;
}
/**
* 讀取文件數(shù)據(jù)
* @param remotePath HDFS文件路徑
* @return 讀取的結(jié)果數(shù)據(jù)
*/
public String readFromFile(String remotePath){
String result = null;
if(fs == null) return null;
try {
FSDataInputStream in = fs.open(new Path(remotePath));
result = in.readUTF();
in.close();
} catch (IOException e) {
e.printStackTrace();
}
return result;
}
/**
* 重命名文件
* @param oldPath 舊文件路徑
* @param newPath 新文件路徑
* @return 是否重命名成功
*/
public boolean renameFile(String oldPath,String newPath){
if(fs == null) return false;
Path old=new Path(oldPath);
Path now=new Path(newPath);
boolean rename = false;
try {
rename = fs.rename(old, now);
} catch (IOException e) {
e.printStackTrace();
}
return rename;
}
/**
* 刪除目錄和文件
* @param remotePath HDFS文件路徑
* @return 是否刪除成功
*/
public boolean deleteFile(String remotePath){
if(fs == null) return false;
boolean success = false;
try {
success = fs.delete(new Path(remotePath), true);
} catch (IOException e) {
e.printStackTrace();
}
return success;
}
/**
* 檢查文件是否存在
* @param remotePath HDFS文件路徑
* @return 是否存在
*/
public boolean existFile(String remotePath){
if(fs == null) return false;
boolean exist = false;
try {
exist = fs.exists(new Path(remotePath));
} catch (IOException e) {
e.printStackTrace();
}
return exist;
}
/**
* 關閉FileSystem
*/
public void closeFileSystem(){
if(fs != null){
try {
fs.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
上面代碼都有注釋，這里就不過多解釋了阁苞，下面來看主類的執(zhí)行代碼：
package com.lzb.hdfs;
import com.lzb.hdfs.fs.HDFSHelper;
public class Demo {
public static void main(String[] args) {
HDFSHelper hdfsHelper = new HDFSHelper();
String dir = "/test";
String filename = "hello.txt";
String path = dir + "/" + filename;
boolean exist = hdfsHelper.existFile(path);
System.out.println(path + " exist file ==> " + exist);
if(!exist){
boolean mkdir = hdfsHelper.mkdir(dir);
System.out.println(dir + " create success ==> " + mkdir);
boolean copyFromLocal = hdfsHelper.copyFromLocal("/"+filename, dir);
System.out.println("upload success ==> " + copyFromLocal);
hdfsHelper.listFiles(dir,false);
String content = "hello world new";
boolean write = hdfsHelper.writeToFile(path, content);
System.out.println("write success ==> " + write);
String data = hdfsHelper.readFromFile(path);
System.out.println("read the data ==> " + data);
String newPath = dir + "/hello2.txt";
boolean renameFile = hdfsHelper.renameFile(path, newPath);
System.out.println("rename success ==> " + renameFile);
boolean copyToLocal = hdfsHelper.copyToLocal(newPath, "/hello2.txt");
System.out.println("download success ==> " + copyToLocal);
//boolean deleteFile = hdfsHelper.deleteFile(newPath);
//System.out.println("delete success ==> " + deleteFile);
}
hdfsHelper.closeFileSystem();
}
}
執(zhí)行結(jié)果如下：
log4j:WARN No appenders could be found for logger (org.apache.hadoop.metrics2.lib.MutableMetricsFactory).
log4j:WARN Please initialize the log4j system properly.
DFS[DFSClient[clientName=DFSClient_NONMAPREDUCE_-1866182384_1, ugi=root (auth:SIMPLE)]]
/test/hello.txt exist file ==> false
/test create success ==> true
upload success ==> true
file path === hdfs://probd/test/hello.txt
file name === hello.txt
file size === 12
file owner === root
file group === supergroup
file permission === rw-r--r--
file blocks === 3
file block size === 134217728
file block nums === 1
block host === Probd01
block host === Probd03
block host === Probd02
write success ==> true
read the data ==> hello world new
rename success ==> true
download success ==> true
