java操作hdfs常用方法封装

最近一段时间又开始做hadoop相关开发，包括hdfs、mr、hbase、hive等，又有了更进一步的了解，下边就简单写些封装好的java操作hdfs的工具类中的方法。

1、从HDFS上指定的文件中读取子目录

public static List<String> getSubDirs(String rootDir) {
        // 存放最后返回的文件夹列表
        List<String> fileList = new ArrayList<String>();
        try {
            FileSystem fs = FileSystem.get(URI.create(rootDir),
                    ConfigurationUtil.conf);
            Path rootPath = new Path(rootDir);
 
            FileStatus fileStatus = fs.getFileStatus(rootPath);
            // 如果传入的是个文件则不再遍历之
            if (!fileStatus.isDir()) {
                fileList.add(fileStatus.getPath().toString());
            } else {
                FileStatus[] fileStatusArray = fs.listStatus(rootPath);
                for (FileStatus fileState : fileStatusArray) {
                    fileList.add(fileState.getPath().toString());
                }
            }
//          fs.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return fileList;
    }

2、从HDFS上读取文件,并存储到本地

public static void readFromHdfs(String fromFile,String outputFile) throws Exception {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(fromFile), conf);
        FSDataInputStream hdfsInStream = fs.open(new Path(fromFile));
 
        OutputStream out = new FileOutputStream(outputFile);
        byte[] ioBuffer = new byte[1024];
        int readLen = hdfsInStream.read(ioBuffer);
        while (-1 != readLen) {
            out.write(ioBuffer, 0, readLen);
            System.out.println(new String(ioBuffer, "utf-8"));
            readLen = hdfsInStream.read(ioBuffer);
        }
        out.close();
        hdfsInStream.close();
//      fs.close();
    }

3、将指定的文件路径从hdfs读取并转换为byte array

public static byte[] readFromFileToByteArray(String srcFile)
            throws Exception {
        if (StringOperatorUtil.isBlank(srcFile)) {
            throw new Exception("所要读取的源文件" + srcFile + ",不存在，请检查!");
        }
        FileSystem fs = FileSystem.get(ConfigurationUtil.conf);
        FSDataInputStream hdfsInStream = fs.open(new Path(srcFile));
 
        byte[] byteArray = new byte[65536];
 
        // 存放最后的所有字节数组
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
 
        // 实际读过来多少
        int readLen = 0;
        while ((readLen = hdfsInStream.read(byteArray)) > 0) {
            bos.write(byteArray);
            byteArray = new byte[65536];
        }
 
        // logger.info("srcFile---"+srcFile+"读到的字节数为---"+readLen);
//      System.out.println("srcFile---" + srcFile + "读到的字节数为---" +);
        hdfsInStream.close();
//      fs.close();
 
        return bos.toByteArray();
    }

4、将一个本地文件或文件夹上传到hdfs上 isCheckExist是否检查输出目录的存在性,若检查，则存在时不予执行复盖操作

public static boolean copyLocalFileToHdfs(String fromFile, String toFile,
            Boolean isCheckExist) throws Exception {
        try {
            FileSystem fs = FileSystem.get(ConfigurationUtil.conf);
 
            Path srcPath = new Path(fromFile); // 原路径
            Path dstPath = new Path(toFile); // 目标路径
 
            // 首先判断源路径是否存在，和输出路轻是否存在，则存在抛出异常，防止被误覆盖掉
            if (!new File(fromFile).exists()) {
                throw new Exception("要拷贝源文件" + fromFile + ",不存在，请检查!");
            }
            if (isCheckExist) {
                if (fs.exists(dstPath)) {
                    // throw new Exception("要拷贝到的输出目录或文件已存在" + dstPath +
                    // ",请检查!");
                    logger.info("要粘贴到的目录" + toFile + ",已存在，不再重新粘贴!");
                    return true;
                }
            }
            // 调用文件系统的文件复制函数,前面参数是指是否删除原文件，true为删除，默认为false
            fs.copyFromLocalFile(false, srcPath, dstPath);
//          fs.close();
 
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }