Hadoop 系列（七）—— HDFS Java API
阿新 • • 發佈：2019-08-07
一、 簡介
想要使用 HDFS API，需要匯入依賴 hadoop-client。如果是 CDH 版本的 Hadoop，還需要額外指明其倉庫地址：
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.heibaiying</groupId> <artifactId>hdfs-java-api</artifactId> <version>1.0</version> <properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <hadoop.version>2.6.0-cdh5.15.2</hadoop.version> </properties> <!---配置 CDH 倉庫地址--> <repositories> <repository> <id>cloudera</id> <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url> </repository> </repositories> <dependencies> <!--Hadoop-client--> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>${hadoop.version}</version> </dependency> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.12</version> <scope>test</scope> </dependency> </dependencies> </project>
二、API的使用
2.1 FileSystem
FileSystem 是所有 HDFS 操作的主入口。由於之後的每個單元測試都需要用到它，這裡使用 @Before 註解進行標註。
private static final String HDFS_PATH = "hdfs://192.168.0.106:8020"; private static final String HDFS_USER = "root"; private static FileSystem fileSystem; @Before public void prepare() { try { Configuration configuration = new Configuration(); // 這裡我啟動的是單節點的 Hadoop,所以副本系數設定為 1,預設值為 3 configuration.set("dfs.replication", "1"); fileSystem = FileSystem.get(new URI(HDFS_PATH), configuration, HDFS_USER); } catch (IOException e) { e.printStackTrace(); } catch (InterruptedException e) { e.printStackTrace(); } catch (URISyntaxException e) { e.printStackTrace(); } } @After public void destroy() { fileSystem = null; }
2.2 建立目錄
支援遞迴建立目錄：
@Test
public void mkDir() throws Exception {
fileSystem.mkdirs(new Path("/hdfs-api/test0/"));
}2.3 建立指定許可權的目錄
FsPermission(FsAction u, FsAction g, FsAction o) 的三個引數分別對應：建立者許可權，同組其他使用者許可權，其他使用者許可權，許可權值定義在 FsAction 列舉類中。
@Test public void mkDirWithPermission() throws Exception { fileSystem.mkdirs(new Path("/hdfs-api/test1/"), new FsPermission(FsAction.READ_WRITE, FsAction.READ, FsAction.READ)); }
2.4 建立檔案，並寫入內容
@Test
public void create() throws Exception {
// 如果檔案存在，預設會覆蓋, 可以通過第二個引數進行控制。第三個引數可以控制使用緩衝區的大小
FSDataOutputStream out = fileSystem.create(new Path("/hdfs-api/test/a.txt"),
true, 4096);
out.write("hello hadoop!".getBytes());
out.write("hello spark!".getBytes());
out.write("hello flink!".getBytes());
// 強制將緩衝區中內容刷出
out.flush();
out.close();
}2.5 判斷檔案是否存在
@Test
public void exist() throws Exception {
boolean exists = fileSystem.exists(new Path("/hdfs-api/test/a.txt"));
System.out.println(exists);
}2.6 檢視檔案內容
檢視小文字檔案的內容，直接轉換成字串後輸出：
@Test
public void readToString() throws Exception {
FSDataInputStream inputStream = fileSystem.open(new Path("/hdfs-api/test/a.txt"));
String context = inputStreamToString(inputStream, "utf-8");
System.out.println(context);
}inputStreamToString 是一個自定義方法，程式碼如下：
/**
* 把輸入流轉換為指定編碼的字元
*
* @param inputStream 輸入流
* @param encode 指定編碼型別
*/
private static String inputStreamToString(InputStream inputStream, String encode) {
try {
if (encode == null || ("".equals(encode))) {
encode = "utf-8";
}
BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, encode));
StringBuilder builder = new StringBuilder();
String str = "";
while ((str = reader.readLine()) != null) {
builder.append(str).append("\n");
}
return builder.toString();
} catch (IOException e) {
e.printStackTrace();
}
return null;
}2.7 檔案重新命名
@Test
public void rename() throws Exception {
Path oldPath = new Path("/hdfs-api/test/a.txt");
Path newPath = new Path("/hdfs-api/test/b.txt");
boolean result = fileSystem.rename(oldPath, newPath);
System.out.println(result);
}2.8 刪除目錄或檔案
public void delete() throws Exception {
/*
* 第二個引數代表是否遞迴刪除
* + 如果 path 是一個目錄且遞迴刪除為 true, 則刪除該目錄及其中所有檔案;
* + 如果 path 是一個目錄但遞迴刪除為 false,則會則丟擲異常。
*/
boolean result = fileSystem.delete(new Path("/hdfs-api/test/b.txt"), true);
System.out.println(result);
}2.9 上傳檔案到HDFS
@Test
public void copyFromLocalFile() throws Exception {
// 如果指定的是目錄，則會把目錄及其中的檔案都複製到指定目錄下
Path src = new Path("D:\\BigData-Notes\\notes\\installation");
Path dst = new Path("/hdfs-api/test/");
fileSystem.copyFromLocalFile(src, dst);
}2.10 上傳大檔案並顯示上傳進度
@Test
public void copyFromLocalBigFile() throws Exception {
File file = new File("D:\\kafka.tgz");
final float fileSize = file.length();
InputStream in = new BufferedInputStream(new FileInputStream(file));
FSDataOutputStream out = fileSystem.create(new Path("/hdfs-api/test/kafka5.tgz"),
new Progressable() {
long fileCount = 0;
public void progress() {
fileCount++;
// progress 方法每上傳大約 64KB 的資料後就會被呼叫一次
System.out.println("上傳進度：" + (fileCount * 64 * 1024 / fileSize) * 100 + " %");
}
});
IOUtils.copyBytes(in, out, 4096);
}2.11 從HDFS上下載檔案
@Test
public void copyToLocalFile() throws Exception {
Path src = new Path("/hdfs-api/test/kafka.tgz");
Path dst = new Path("D:\\app\\");
/*
* 第一個引數控制下載完成後是否刪除原始檔,預設是 true,即刪除;
* 最後一個引數表示是否將 RawLocalFileSystem 用作本地檔案系統;
* RawLocalFileSystem 預設為 false,通常情況下可以不設定,
* 但如果你在執行時候丟擲 NullPointerException 異常,則代表你的檔案系統與程式可能存在不相容的情況 (window 下常見),
* 此時可以將 RawLocalFileSystem 設定為 true
*/
fileSystem.copyToLocalFile(false, src, dst, true);
}2.12 檢視指定目錄下所有檔案的資訊
public void listFiles() throws Exception {
FileStatus[] statuses = fileSystem.listStatus(new Path("/hdfs-api"));
for (FileStatus fileStatus : statuses) {
//fileStatus 的 toString 方法被重寫過，直接列印可以看到所有資訊
System.out.println(fileStatus.toString());
}
}FileStatus 中包含了檔案的基本資訊，比如檔案路徑，是否是資料夾，修改時間，訪問時間，所有者，所屬組，檔案許可權，是否是符號連結等，輸出內容示例如下：
FileStatus{
path=hdfs://192.168.0.106:8020/hdfs-api/test;
isDirectory=true;
modification_time=1556680796191;
access_time=0;
owner=root;
group=supergroup;
permission=rwxr-xr-x;
isSymlink=false
}2.13 遞迴檢視指定目錄下所有檔案的資訊
@Test
public void listFilesRecursive() throws Exception {
RemoteIterator<LocatedFileStatus> files = fileSystem.listFiles(new Path("/hbase"), true);
while (files.hasNext()) {
System.out.println(files.next());
}
}和上面輸出類似，只是多了文字大小，副本系數，塊大小資訊。
LocatedFileStatus{
path=hdfs://192.168.0.106:8020/hbase/hbase.version;
isDirectory=false;
length=7;
replication=1;
blocksize=134217728;
modification_time=1554129052916;
access_time=1554902661455;
owner=root; group=supergroup;
permission=rw-r--r--;
isSymlink=false}2.14 檢視檔案的塊資訊
@Test
public void getFileBlockLocations() throws Exception {
FileStatus fileStatus = fileSystem.getFileStatus(new Path("/hdfs-api/test/kafka.tgz"));
BlockLocation[] blocks = fileSystem.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
for (BlockLocation block : blocks) {
System.out.println(block);
}
}塊輸出資訊有三個值，分別是檔案的起始偏移量 (offset)，檔案大小 (length)，塊所在的主機名 (hosts)。
0,57028557,hadoop001這裡我上傳的檔案只有 57M(小於 128M)，且程式中設定了副本系數為 1，所有隻有一個塊資訊。
以上所有測試用例下載地址：HDFS Java API
更多大資料系列文章可以參見 GitHub 開源專案： 大資料入門指南