Hadoop編程實現之HDFS

HDFS原理圖：

下面我們來寫一個基於HDFS的demo，該demo主要實現的是將HDFS上的一個文件內容讀取出來並保存到另一個文件上的功能。

1.輔助類

這個類主要是用來獲取hdfs文件系統連接的

public class HdfsUtils {
    
    /**
     * @return
     * @throws Exception
     */
    public static FileSystem getFileSystem() throws Exception{
        
        Configuration conf = new Configuration() ;
        conf.set("fs.defaultFS", "hdfs://192.168.1.109:8020");
        
        FileSystem fileSystem = FileSystem.get(conf) ;
        
        return fileSystem ;
    }
    /**
     * @param pOpenUri
     * @param pUser
     * @return
     * @throws Exception
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    public static FileSystem getFileSystemByUser(String pOpenUri,String pUser) throws Exception, InterruptedException, URISyntaxException{
        
        Configuration conf = new Configuration() ;
        conf.set("fs.defaultFS", "hdfs://192.168.1.109:8020");
        
        FileSystem fileSystem = FileSystem.get(new URI(pOpenUri), conf, pUser) ;
        
        return fileSystem ;
        
    }
    
    /**
     * @param pUser
     * @return
     * @throws Exception
     * @throws InterruptedException
     * @throws URISyntaxException
     */
    public static FileSystem getFileSystemByUser(String pUser) throws Exception, InterruptedException, URISyntaxException{
        
        String fileUri = "/home/test/test.txt" ;
        
        Configuration conf = new Configuration() ;
        conf.set("fs.defaultFS", "hdfs://192.168.1.109:8020");
        
        FileSystem fileSystem = FileSystem.get(new URI(fileUri), conf, pUser) ;
        
        return fileSystem ;
        
    }
    
    

}

2.主類

這個類主要是用來進行文件讀寫和創建的

public class HdfsFsTest {

    public static void main(String[] args) {

        String fileUri = "/home/test/test.txt";
        String fileOutputUrl = "/home/test/out.txt";
        try {

            writeFileToHdfs(fileUri, fileOutputUrl);
            System.out.println("DONE!");

        } catch (Exception e) {

            e.printStackTrace();
        }

    }

    public static void writeFileToHdfs(String pOpenUri, String pOutputUrl)
            throws Exception {

        FileSystem fileSystem = null;
        FSDataInputStream fileInputStream = null;
        FSDataOutputStream fileOutputStream = null;
        int buffSize = 4096;

        try {

            fileSystem = HdfsUtils.getFileSystem();

            fileInputStream = fileSystem.open(new Path(pOpenUri));
            fileOutputStream = fileSystem.create(new Path(pOutputUrl));
            IOUtils.copyBytes(fileInputStream, fileOutputStream, buffSize,
                    false);

        } catch (Exception e) {
            throw e;
        } finally {
            IOUtils.closeStream(fileInputStream);

            IOUtils.closeStream(fileOutputStream);
        }

    }

}

3.運行結果

運行成功！