Java操作HDFS示例

1. 環(huán)境準(zhǔn)備

  1. 大數(shù)據(jù)集群一套，沒有的可以自己本地搭建一套（參考地址：http://www.itdecent.cn/p/2c2ae6490fa0）
  2. 本地安裝JDK
  3. 本地安裝IDEA或者Eclipse

2. 樣例代碼：

package com.lancer.hdfs;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.io.IOUtils;
import org.junit.Test;

public class HDFSClient {


    /**
     * 創(chuàng)建目錄
     * @throws Exception
     * @throws IOException
     * @throws URISyntaxException
     */
    @Test
    public void mkdir() throws Exception,IOException,URISyntaxException {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://hadoop102:9000");
        // 獲取hdfs客戶端對象
        //FileSystem fs = FileSystem.get(conf); 
        FileSystem fs = FileSystem.get(new java.net.URI("hdfs://hadoop102:9000")  , conf,  "root");
        
        // 在hdfs上創(chuàng)建路徑
        fs.mkdirs(new Path("/client/test4"));

        // 關(guān)閉資源
        fs.close();
        
        System.out.println("done");
    }
    
    
    /**
     * 上傳文件
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void testCopyFromLocalFile() throws IOException, InterruptedException, URISyntaxException {
        // 1. 獲取fs對象
        FileSystem fs = FileSystem.get(new java.net.URI("hdfs://hadoop102:9000")  , new Configuration(),  "root");
        //2 執(zhí)行上傳API
        fs.copyFromLocalFile(new Path("D:/學(xué)習(xí)/Hadoop權(quán)威指南（第四版）.pdf"), new Path("/client/test"));
        // 3 關(guān)閉資源
        fs.close();
    }
    
    /**
     * 上傳文件，并設(shè)置備份數(shù)量
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void testCopyFromLocalFile2() throws IOException, InterruptedException, URISyntaxException {

            // 1 獲取文件系統(tǒng)
            Configuration configuration = new Configuration();
            //configuration.set("dfs.replication", "2");
            FileSystem fs = FileSystem.get(new java.net.URI("hdfs://hadoop102:9000"), configuration, "root");

            // 2 上傳文件
            fs.copyFromLocalFile(new Path("D:/學(xué)習(xí)/Hadoop權(quán)威指南（第四版）.pdf"), new Path("/client/test3"));

            // 3 關(guān)閉資源
            fs.close();

            System.out.println("done");
    }

    /**
     * 文件下載
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void testCopyToLocalFile() throws IOException, InterruptedException, URISyntaxException{

            // 1 獲取文件系統(tǒng)
            Configuration configuration = new Configuration();
            FileSystem fs = FileSystem.get(new java.net.URI("hdfs://hadoop102:9000"), configuration, "root");
            
            // 2 執(zhí)行下載操作
            // boolean delSrc 指是否將原文件刪除
            // Path src 指要下載的文件路徑
            // Path dst 指將文件下載到的路徑
            // boolean useRawLocalFileSystem 是否開啟文件校驗
            fs.copyToLocalFile(false, new Path("/client/test3/Hadoop權(quán)威指南（第四版）.pdf"), new Path("d:/Hadoop權(quán)威指南（第四版）.pdf"), true);
            
            // 3 關(guān)閉資源
            fs.close();
            System.out.println("done");
    }

    
    /**
     * 文件夾刪除
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void testDelete() throws IOException, InterruptedException, URISyntaxException{

        // 1 獲取文件系統(tǒng)
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new java.net.URI("hdfs://hadoop102:9000"), configuration, "root");
            
        // 2 執(zhí)行刪除
        fs.delete(new Path("/client/"), true);
            
        // 3 關(guān)閉資源
        fs.close();
        
        System.out.println("done");
    }
    
    /**
     * 文件名修改
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void testRename() throws IOException, InterruptedException, URISyntaxException{

        // 1 獲取文件系統(tǒng)
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new java.net.URI("hdfs://hadoop102:9000"), configuration, "root"); 
            
        // 2 修改文件名稱
        fs.rename(new Path("/README.txt"), new Path("/README222.txt"));
            
        // 3 關(guān)閉資源
        fs.close();
        
        System.out.println("done");
    }

    
    @Test
    public void testListFiles() throws IOException, InterruptedException, URISyntaxException{

        // 1獲取文件系統(tǒng)
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new java.net.URI("hdfs://hadoop102:9000"), configuration, "root"); 
            
        // 2 獲取文件詳情
        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);
            
        while(listFiles.hasNext()){
            LocatedFileStatus status = listFiles.next();
                
            // 輸出詳情
            // 文件名稱
            System.out.println("文件名：" + status.getPath().getName());
            // 長度
            System.out.println("長度：" + status.getLen());
            // 權(quán)限
            System.out.println("權(quán)限：" + status.getPermission());
            // 分組
            System.out.println("分組：" + status.getGroup());
                
            // 獲取存儲的塊信息
            BlockLocation[] blockLocations = status.getBlockLocations();
            System.out.println("存儲的塊信息：" + status.getGroup());
            for (BlockLocation blockLocation : blockLocations) {
                    
                // 獲取塊存儲的主機節(jié)點
                String[] hosts = blockLocation.getHosts();
                    
                for (String host : hosts) {
                    System.out.println(host);
                }
            }
                
            System.out.println("-----------班長的分割線----------");
        }

        // 3 關(guān)閉資源
        fs.close();
        
        System.out.println("done");
    }

    
    /**
     * HDFS文件和文件夾判斷
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void testListStatus() throws IOException, InterruptedException, URISyntaxException{
            
        // 1 獲取文件配置信息
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new java.net.URI("hdfs://hadoop102:9000"), configuration, "root");
            
        // 2 判斷是文件還是文件夾
        FileStatus[] listStatus = fs.listStatus(new Path("/"));
            
        for (FileStatus fileStatus : listStatus) {
            
            // 如果是文件
            if (fileStatus.isFile()) {
                    System.out.println("f:"+fileStatus.getPath().getName());
                }else {
                    System.out.println("d:"+fileStatus.getPath().getName());
                }
            }
            
        // 3 關(guān)閉資源
        fs.close();
        
        System.out.println("done");
    }

    
    /**
     * IO操作，上傳文件，不使用封裝好的方法
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void putFileToHDFS() throws IOException, InterruptedException, URISyntaxException {

        // 1 獲取文件系統(tǒng)
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "root");

        // 2 創(chuàng)建輸入流
        FileInputStream fis = new FileInputStream(new File("D:/學(xué)習(xí)/Hadoop權(quán)威指南（第四版）.pdf"));

        // 3 獲取輸出流
        FSDataOutputStream fos = fs.create(new Path("/Hadoop權(quán)威指南（第四版）.pdf"));

        // 4 流對拷
        IOUtils.copyBytes(fis, fos, configuration);

        // 5 關(guān)閉資源
        IOUtils.closeStream(fos);
        IOUtils.closeStream(fis);
        fs.close();
        
        System.out.println("done");
    }

    
    /**
     * O操作，文件下載，不使用封裝好的方法
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void getFileFromHDFS() throws IOException, InterruptedException, URISyntaxException{

        // 1 獲取文件系統(tǒng)
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "root");
            
        // 2 獲取輸入流
        FSDataInputStream fis = fs.open(new Path("/Hadoop權(quán)威指南（第四版）.pdf"));
            
        // 3 獲取輸出流
        FileOutputStream fos = new FileOutputStream(new File("d:/Hadoop權(quán)威指南（第四版）222.pdf"));
            
        // 4 流的對拷
        IOUtils.copyBytes(fis, fos, configuration);
            
        // 5 關(guān)閉資源
        IOUtils.closeStream(fos);
        IOUtils.closeStream(fis);
        fs.close();
        
        System.out.println("done");
    }
    
    /**
     * 分塊下載文件，第一塊
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void readFileSeek1() throws IOException, InterruptedException, URISyntaxException{

        // 1 獲取文件系統(tǒng)
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "root");
            
        // 2 獲取輸入流
        FSDataInputStream fis = fs.open(new Path("/hadoop-2.7.2.tar.gz"));
            
        // 3 創(chuàng)建輸出流
        FileOutputStream fos = new FileOutputStream(new File("d:/hadoop-2.7.2.tar.gz.part1"));
            
        // 4 流的拷貝
        byte[] buf = new byte[1024];
            
        for(int i =0 ; i < 1024 * 128; i++){
            fis.read(buf);
            fos.write(buf);
        }
            
        // 5關(guān)閉資源
        IOUtils.closeStream(fis);
        IOUtils.closeStream(fos);
        fs.close();
        
        System.out.println("done");
    }

    
    /**
     * 分塊下載文件，第二塊
     * @throws IOException
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    @Test
    public void readFileSeek2() throws IOException, InterruptedException, URISyntaxException{

        // 1 獲取文件系統(tǒng)
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://hadoop102:9000"), configuration, "root");
            
        // 2 打開輸入流
        FSDataInputStream fis = fs.open(new Path("/hadoop-2.7.2.tar.gz"));
            
        // 3 定位輸入數(shù)據(jù)位置
        fis.seek(1024*1024*128);
            
        // 4 創(chuàng)建輸出流
        FileOutputStream fos = new FileOutputStream(new File("d:/hadoop-2.7.2.tar.gz.part2"));
            
        // 5 流的對拷
        IOUtils.copyBytes(fis, fos, configuration);
            
        // 6 關(guān)閉資源
        IOUtils.closeStream(fis);
        IOUtils.closeStream(fos);
        System.out.println("done");
        
        /**
         * 
         * 在Window命令窗口中進(jìn)入到目錄E:\，然后執(zhí)行如下命令，對數(shù)據(jù)進(jìn)行合并
         * type hadoop-2.7.2.tar.gz.part2 >> hadoop-2.7.2.tar.gz.part1
         * 合并完成后，將hadoop-2.7.2.tar.gz.part1重新命名為hadoop-2.7.2.tar.gz。解壓發(fā)現(xiàn)該tar包非常完整。
         */
    }



}
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容