Hadoop 系列（七）—— HDFS Java API
- 2019 年 10 月 3 日
- 筆記
一、 簡介
想要使用 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 開源項目： 大數據入門指南
