HDFS文件系统的操作

标签：

package com.bank.utils;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.Date;

import org.apache.hadoop.conf.Configuration;
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.Path;
import org.apache.hadoop.io.IOUtils;

/**
 * HDFS常规操作
 * @author mengyao
 *
 */
public class HDFSUtils {

    private final static String DFS_PATH = "hdfs://ns1";
    private final static String USER = "root";
    
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", DFS_PATH);
        conf.set("dfs.nameservices", "ns1");
        conf.set("dfs.ha.namenodes.ns1", "nn1,nn2");
        conf.set("dfs.namenode.rpc-address.ns1.nn1", "h1:9000");
        conf.set("dfs.namenode.rpc-address.ns1.nn2", "h2:9000");
        conf.set("dfs.client.failover.proxy.provider.ns1", "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");

        FileSystem fs = FileSystem.get(new URI("hdfs://ns1"), conf, USER);
        
        //在HDFS上创建文件夹
        createDir(fs, "/hkd/hongkong");
        
        //删除HDFS上的文件夹或文件，文件夹为true
        deleteFileOrDir(fs, "/hkd");
        
        //上传本地文件到HDFS上，如果文件存在则覆盖
        upload(fs, "D:/data", "/data.dat");
        
        //从HDFS上下载文件到本地
        download(fs, "/cny/data/data", "D:/data");
        
        //删除HDFS上的文件，如果存在
        deleteFile(fs, "/data.dat");
        
        //读取HDFS上指定目录下的所有文件及文件夹信息
        readDfsPath(fs, "/cny");
    }
    
    public static boolean createDir(FileSystem fs, String dfsNewDir){
        boolean status = false;
        try {
            if (fs.exists(new Path(dfsNewDir))) {
                System.err.println(" this dir exist !");
                return status;
            }
            status = fs.mkdirs(new Path(dfsNewDir));
        } catch (IllegalArgumentException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                fs.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        
        return status;
    }
    
    public static boolean deleteFileOrDir(FileSystem fs, String dfsPath){
        boolean status = false;
        try {
            status = fs.delete(new Path(dfsPath), true);
        } catch (IllegalArgumentException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                fs.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        
        return status;
    }
    
    public static boolean upload(FileSystem fs, String localPath, String dfsPath){
        boolean status = false;
        try {
            FSDataOutputStream out = fs.create(new Path(dfsPath), true);
            BufferedInputStream in = new BufferedInputStream(new FileInputStream(new File(localPath)));
            IOUtils.copyBytes(in, out, 4096, true);
        } catch (IllegalArgumentException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                fs.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        
        return status;
    }
    
    public static boolean download(FileSystem fs, String dfsPath, String localPath){
        boolean status = false;
        try {
            FSDataInputStream in = fs.open(new Path(dfsPath));
            BufferedOutputStream out = new BufferedOutputStream(new FileOutputStream(new File(localPath)));
            IOUtils.copyBytes(in, out, 4096, true);
            status = true;
        } catch (IllegalArgumentException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                fs.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        
        return status;
    }
    
    public static boolean deleteFile(FileSystem fs, String dfsPath){
        boolean status = false;
        try {
            if (fs.exists(new Path(dfsPath))) {
                status = fs.delete(new Path(dfsPath), true);                
            }
        } catch (IllegalArgumentException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                fs.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        
        return status;
    }
    
    
    public static void readDfsPath(FileSystem fs, String dfsPath){
        try {
            FileStatus[] listStatus = fs.listStatus(new Path(dfsPath));
            for (FileStatus fsStat : listStatus) {
                String isDir = fsStat.isDirectory()?"文件夹":"文件";  
                final String permission = fsStat.getPermission().toString();  
                final short replication = fsStat.getReplication();  
                final long len = fsStat.getLen();  
                final String dateStr = new SimpleDateFormat("yyyy-MM-dd hh:MM:ss").format(new Date(fsStat.getAccessTime()));  
                final String path = fsStat.getPath().toString();  
                System.out.println(isDir+"\t"+permission+"\t"+replication+"\t"+len+"\t"+dateStr+"\t"+path);  
            }
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IllegalArgumentException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

HDFS文件系统的操作

标签：

原文地址：http://www.cnblogs.com/mengyao/p/4234955.html