使用Junit封裝HFDS

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.net.URI;
/**
 * 使用Java API操作HDFS文件系統
 */
public class HDFSApp {
    public static final String HDFS_PATH = "hdfs://localhost:9000";
    Configuration conf = null;
    FileSystem fs = null;

    @Before
    public void open() throws Exception{
        System.out.println("連接HDFS...");
        conf = new Configuration();
        // 副本系數為1,配置文件只對shell生效
        conf.set("dfs.replication","1");
        /**
         * 構造一個訪問指定HDFS文件系統的客戶端對象
         * 第一個參數： HDFS的URI
         * 第二個參數: 客戶端指定的配置參賽
         * 第三個參數： 客戶端的身份,說白了就是用戶名
         */
        fs = FileSystem.get(new URI(HDFS_PATH),conf,"hadoop");
    }
    @After
    public void close() throws Exception{
        conf = null;
        fs = null;
        System.out.println("註銷連接...");
    }
}
 

創建HDFS文件夾

/**
 * 創建HDFS文件夾
 */
@Test
public void mkdir() throws Exception{
    fs.mkdirs(new Path("input"));
}

文件上傳

/**
 * 文件上傳
 */
@Test
public void copyFromLocalFile() throws Exception{
    fs.copyFromLocalFile(new Path("/home/hadoop/word.txt"),
                         new Path("input/word.txt"));
}
 

文件下載

/**
 * 文件下載
 */
@Test
public void copyToLocalFile() throws Exception{
    fs.copyToLocalFile(new Path("input/word.txt"),
                       new Path("/home/hadoop/word2.txt"));
}

查看HDFS文件內容

/**
 * 查看HDFS文件內容
 */
@Test
public void catFileText() throws Exception{
    FSDataInputStream in = fs.open(new Path("input/word.txt"));
    IOUtils.copyBytes(in,System.out,1024);
}
 

列出指定文件夾下的所有內容

/**
 * 列出指定文件夾下的所有內容
 */
@Test
public void listFile() throws Exception{
    FileStatus[] listStatus = fs.listStatus(new Path("input"));
    for (FileStatus file : listStatus) {
        String isDir = file.isDirectory()?"文件夾":"文件";
        String permission = file.getPermission().toString();
        short replication = file.getReplication();
        long len = file.getLen();
        String path = file.getPath().toString();
        // 輸出信息
        System.out.println(isDir+"\t"+permission+"\t"+
                           replication+"\t"+len+"\t"+path);
    }
}

遞歸列出指定文件夾下的所有文件(夾)信息

/**
 * 遞歸列出指定文件夾下的所有文件(夾)信息
 */
@Test
public void listAllFiles() throws Exception{
    RemoteIterator<LocatedFileStatus> files = fs.listFiles(new Path("input"),true);
    while (files.hasNext()){
        LocatedFileStatus file = files.next();
        // 獲取信息
        String isDir = file.isDirectory()?"文件夾":"文件";
        String permission = file.getPermission().toString();
        short replication = file.getReplication();
        long len = file.getLen();
        String path = file.getPath().toString();
        // 輸出信息
        System.out.println(isDir+"\t"+permission+"\t"+
                           replication+"\t"+len+"\t"+path);
    }
}

創建HDFS文件,並寫入內容

/**
 * 創建HDFS文件,並寫入內容
 */
@Test
public void create() throws Exception{
    FSDataOutputStream out = fs.create(new Path("input/a.txt"));
    out.writeUTF("Hello,HDFS!");
    out.flush();
    out.close();
}

刪除文件/文件夾

/**
 * 刪除文件/文件夾
 */
@Test
public void deleteFile() throws Exception{
    // true遞歸刪除文件夾,false不刪除文件夾,文件則無所謂
    fs.delete(new Path("input"),true);
}

HDFS重命名

/**
 * HDFS重命名
 */
@Test
public void rename() throws Exception{
    fs.rename(new Path("input/word.txt"),new Path("input/input.txt"));
}

列出文件塊信息

/**
 * 列出文件塊信息
 */
@Test
public void getFileBlockLocations() throws Exception{
    FileStatus fileStatus = fs.getFileStatus(new Path("input/word.txt"));
    BlockLocation[] blocks = fs.getFileBlockLocations(fileStatus, 0,
                                                      fileStatus.getLen());
    for (BlockLocation block : blocks) {
        // 獲取文件塊名字(多個,被切分)
        for (String name:block.getNames()) {
            System.out.println(name+":"+block.getOffset()+":"+block.getLength());
        }
    }
}

使用Java API操作HDFS文件系統