HDFS-JavaAPI

1、pom.xml

<?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>yofc</groupId>
    <artifactId>root</artifactId>
    <version>1.0-SNAPSHOT</version>

    <dependencies>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>2.9.2</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <!-- 指定jdk -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.0</version>
                <configuration>
                    <encoding>UTF-8</encoding>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

 

2、測試

import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.log4j.BasicConfigurator; import org.junit.After; import org.junit.Before; import org.junit.Test; import java.io.IOException; import java.net.URI; public class HDFSClient { private Configuration conf; private FileSystem fs; @Before public void init() throws Exception { // 設置 HADOOP_HOME 環境變量
        System.setProperty("hadoop.home.dir", "D:/DevelopTools/hadoop-2.9.2/"); // 日誌初始化
 BasicConfigurator.configure(); conf = new Configuration(); //conf.set("fs.defaultFS", "hdfs://192.168.8.136:9000"); //fs = FileSystem.get(conf ); // 獲取 hdfs 客戶端對象，指定用戶名，避免無權限
        fs = FileSystem.get(new URI("hdfs://192.168.8.136:9000"), conf, "root"); } @After public void close() throws IOException { fs.close(); } // 在 hdfs 上建立文件夾
 @Test public void mkdirs() throws IOException { fs.mkdirs(new Path("/10086/")); } }

 

文件上傳

@Test public void testCopyFromLocalFile() throws Exception{ fs.copyFromLocalFile(new Path("D:/MyFile/Downloads/Writage-1.12.msi"), new Path("/Writage-1.12.msi")); }

手動 IO 流方式

@Test public void putFileToHDFS() throws Exception{ // 獲取輸入流
    FileInputStream fis = new FileInputStream(new File("D://MyFile/Downloads/Writage-1.12.msi")); // 獲取輸出流
    FSDataOutputStream fos = fs.create(new Path("/Writage-1.12.msi")); // 流的對拷
 IOUtils.copyBytes(fis, fos, conf); // 關閉資源
 IOUtils.closeStream(fos); IOUtils.closeStream(fis); }

 

 

文件下載

@Test public void testCopyToLocalFile() throws Exception{ // fs.copyToLocalFile(new Path("/AAA.txt"), new Path("d:/BBB.txt"));
    /** * delSrc：是否刪除原數據 * src：hdfs 路徑 * dst：本地 路徑 * useRawLocalFileSystem：crc 文件完整性校驗 */ fs.copyToLocalFile(false, new Path("/Writage-1.12.msi"), new Path("D://Writage.msi"), true); }

手動 IO 流方式

@Test public void getFileFromHDFS() throws Exception{ // 獲取輸入流
    FSDataInputStream fis = fs.open(new Path("/Writage-1.12.msi")); // 獲取輸出流
    FileOutputStream fos = new FileOutputStream(new File("D://Writage.msi")); // 流的對拷
 IOUtils.copyBytes(fis, fos, conf); // 關閉資源
 IOUtils.closeStream(fos); IOUtils.closeStream(fis); }

分塊方式，這裏要下載的文件被 hdfs 切割成了 3 塊

// 下載第一塊
@Test public void readFileSeek1() throws Exception{ // 獲取輸入流
    FSDataInputStream fis = fs.open(new Path("/hadoop-2.9.2-win10-64.tar.gz")); // 獲取輸出流
    FileOutputStream fos = new FileOutputStream(new File("D://hadoop-2.9.2-win10-64.tar.gz")); // 流的對拷（只拷貝128m）
    byte[] buf = new byte[1024]; for (int i = 0; i < 1024 * 128; i++) { fis.read(buf); fos.write(buf); } // 關閉資源
 IOUtils.closeStream(fos); IOUtils.closeStream(fis); } // 下載第二塊
@Test public void readFileSeek2() throws Exception{ // 獲取輸入流
    FSDataInputStream fis = fs.open(new Path("/hadoop-2.9.2-win10-64.tar.gz")); // 設置指定讀取的起點
    fis.seek(1024*1024*128); // 獲取輸出流
    FileOutputStream fos = new FileOutputStream(new File("D://hadoop-2.9.2-win10-64.tar.gz2")); // 流的對拷（只拷貝128m）
    byte[] buf = new byte[1024]; for (int i = 0; i < 1024 * 128; i++) { fis.read(buf); fos.write(buf); } // 關閉資源
 IOUtils.closeStream(fos); IOUtils.closeStream(fis); } // 下載第三塊
@Test public void readFileSeek3() throws Exception{ // 獲取輸入流
    FSDataInputStream fis = fs.open(new Path("/hadoop-2.9.2-win10-64.tar.gz")); // 設置指定讀取的起點
    fis.seek(1024*1024*128*2); // 獲取輸出流
    FileOutputStream fos = new FileOutputStream(new File("D://hadoop-2.9.2-win10-64.tar.gz3")); // 流的對拷
 IOUtils.copyBytes(fis, fos, conf); // 關閉資源
 IOUtils.closeStream(fos); IOUtils.closeStream(fis); }

分塊下載完畢後合併文件

# Windows 環境下

# 將 hadoop-2.9.2-win10-64.tar.gz2 追加到 hadoop-2.9.2-win10-64.tar.gz
type hadoop-2.9.2-win10-64.tar.gz2 >> hadoop-2.9.2-win10-64.tar.gz # 將 hadoop-2.9.2-win10-64.tar.gz3 追加到 hadoop-2.9.2-win10-64.tar.gz
type hadoop-2.9.2-win10-64.tar.gz3 >> hadoop-2.9.2-win10-64.tar.gz # 最後 hadoop-2.9.2-win10-64.tar.gz 就是一個完整的文件了

 

 

文件刪除

@Test public void testDelete() throws Exception{ /** * var1：hdfs 路徑 * var2：是否遞歸刪除，若爲文件夾則必須爲 true */ fs.delete(new Path("/Writage-1.12.msi"), true); }

 

重命名

@Test public void testRename() throws Exception{ // 把根目錄的 10086 改成 mkmk
    fs.rename(new Path("/10086/"), new Path("/mkmk/")); }

 

查看文件詳情

@Test public void testListFiles() throws Exception{ // 遞歸獲取根目錄下的全部文件
    RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true); while(listFiles.hasNext()){ LocatedFileStatus fileStatus = listFiles.next(); // 文件名稱
 System.out.println(fileStatus.getPath().getName()); // 文件權限
 System.out.println(fileStatus.getPermission()); // 文件長度
 System.out.println(fileStatus.getLen()); // 文件塊信息
        BlockLocation[] blockLocations = fileStatus.getBlockLocations(); for (BlockLocation blockLocation : blockLocations) { // 文件塊所在的主機名
            String[] hosts = blockLocation.getHosts(); for (String host : hosts) { System.out.println(host); } } System.out.println("-------------------"); } }

 

判斷是文件仍是文件夾

@Test public void testListStatus() throws Exception{ FileStatus[] listStatus = fs.listStatus(new Path("/")); for (FileStatus fileStatus : listStatus) { if (fileStatus.isFile()) { System.out.println("文件："+fileStatus.getPath().getName()); }else{ System.out.println("文件夾："+fileStatus.getPath().getName()); } } }

 

 

---

Windows 運行 Hadoop 問題：https://wiki.apache.org/hadoop/WindowsProblems