spark读取压缩文件

spark读取压缩文件，对同一个压缩文件内文件进行分布式处理，粒度：文件级

-| .rar.gz

   -| .gz

   -| .zip

-| .zip　　

   -| .gz

   -| .zip

使用 sc.binaryFile()得到-> JavaPairRDD<String,PortableDataStream>

key是压缩文件根目录，PortableDataStream是根目录的二进制流。

并行化处理：将每个压缩文件根据内部文件拆分成文件流，实现1：n的并行度

// 一个压缩包流，对应多个流，每个流对应一个文件名称
    public static JavaPairRDD<PortableDataStream, FilePropertyPojo> getFileListRdd(
            JavaPairRDD<String, PortableDataStream> zipRdd) {
        return zipRdd.flatMapToPair(tuple2 -> {
            List<Tuple2<PortableDataStream, FilePropertyPojo>> targetList = new ArrayList<>();
            List<FilePropertyPojo> fileNameList = getFileNameList(tuple2._2);

            for (FilePropertyPojo filePropertyPojo : fileNameList) {

                targetList.add(new Tuple2<>(tuple2._2, filePropertyPojo));
            }
            return targetList.iterator();

        });

    }

    private static List<FilePropertyPojo> getFileNameList(PortableDataStream portableDataStream) {
        List<FilePropertyPojo> fileNameList = new ArrayList<>();
        try {
            List<FilePropertyPojo> mrPropertyPojoList = new ArrayList<>();
            String path = portableDataStream.getPath();

            String fileCompressMode = path.substring(path.lastIndexOf('.')).toLowerCase();
            switch (fileCompressMode) {
            case ".gz":
                getFileNameFromGz(portableDataStream, mrPropertyPojoList);
                break;
            case ".zip":
                getFileNameFromZip(portableDataStream, mrPropertyPojoList);
                break;

            default:
            }
            return mrPropertyPojoList;

        } catch (Exception e) {
            //
        }
        return fileNameList;
    }

    private static void getFileNameFromGz(PortableDataStream portableDataStream,
            List<FilePropertyPojo> mrPropertyPojoList) {
        try (TarArchiveInputStream inputStream = new TarArchiveInputStream(
                new GZIPInputStream(portableDataStream.open()))) {
            TarArchiveEntry tarArchiveEntry;
            while ((tarArchiveEntry = inputStream.getNextTarEntry()) != null) {
                try {
                    getEachFileName(mrPropertyPojoList, tarArchiveEntry.getName(), tarArchiveEntry.getSize());

                } catch (Exception e) {
                    //
                }
            }
        } catch (Exception e) {
            //
        }
    }

    private static void getFileNameFromZip(PortableDataStream portableDataStream,
            List<FilePropertyPojo> mrPropertyPojoList) throws IOException {

        try (ZipArchiveInputStream zipArchiveInputStream = new ZipArchiveInputStream(portableDataStream.open())) {
            ZipArchiveEntry nextZipEntry;
            while ((nextZipEntry = zipArchiveInputStream.getNextZipEntry()) != null) {
                try {
                    getEachFileName(mrPropertyPojoList, nextZipEntry.getName(), nextZipEntry.getSize());

                } catch (Exception e) {
                    //
                }
            }
        }
    }