Java에서 파일을 읽고 ZIP으로 처리하기

파일 접근에는 저수준 변환의 FoundryInput과 FoundryOutput을 사용합니다. 입력은 asFiles().getFileSystem(), 출력은 getFileSystem()으로 파일시스템을 얻습니다. 인자 없는 출력은 스냅샷으로 쓰며 증분 파일 처리에는 쓰기 모드를 별도로 지정해야 합니다.

아래 클래스들은 수동 등록용입니다. 파일을 처리하는 단위와 결과 경로를 구분하여 설계합니다.

ZIP 파일별 병렬 해제

Spark는 입력 ZIP 파일마다 작업을 나눌 수 있지만 ZIP 한 개의 압축 스트림 자체는 분할 처리하지 않습니다. 한 압축 파일 내부의 분산 처리가 필요하면 해당 reader가 지원하는 splittable 형식, 예를 들어 bzip2를 검토합니다.

다음 myproject/files/UnzipWithSpark.java는 각 입력 ZIP 안의 파일을 출력 파일시스템에 씁니다. 출력 경로 앞에 원래 ZIP의 논리 경로를 붙여 서로 다른 압축 파일 안의 같은 이름을 구분합니다.

package myproject.files;

import com.google.common.io.ByteStreams;
import com.palantir.transforms.lang.java.api.Compute;
import com.palantir.transforms.lang.java.api.FoundryInput;
import com.palantir.transforms.lang.java.api.FoundryOutput;
import com.palantir.transforms.lang.java.api.ReadOnlyLogicalFileSystem;
import com.palantir.transforms.lang.java.api.WriteOnlyLogicalFileSystem;
import com.palantir.util.syntacticpath.Paths;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.zip.ZipEntry;
import java.util.zip.ZipInputStream;

public final class UnzipWithSpark {
    @Compute
    public void compute(FoundryInput zipFiles, FoundryOutput output) {
        ReadOnlyLogicalFileSystem source = zipFiles.asFiles().getFileSystem();
        WriteOnlyLogicalFileSystem destination = output.getFileSystem();

        source.filesAsDataset().foreach(portableFile ->
                portableFile.processWithThenClose(stream -> {
                    try (ZipInputStream zip = new ZipInputStream(stream)) {
                        ZipEntry entry;
                        while ((entry = zip.getNextEntry()) != null) {
                            if (!entry.isDirectory()) {
                                String name = entry.getName();
                                requireRelativeEntry(name);
                                destination.writeTo(
                                        Paths.get(portableFile.getLogicalPath().toString(), name),
                                        target -> ByteStreams.copy(zip, target));
                            }
                            zip.closeEntry();
                        }
                        return null;
                    } catch (IOException exception) {
                        throw new UncheckedIOException(exception);
                    }
                }));
    }

    private static void requireRelativeEntry(String name) {
        if (name.isEmpty() || name.startsWith("/") || name.indexOf('\\') >= 0
                || (name.length() > 1 && name.charAt(1) == ':')) {
            throw new IllegalArgumentException("ZIP entry must use a relative path: " + name);
        }
        int start = 0;
        for (int index = 0; index <= name.length(); index++) {
            if (index == name.length() || name.charAt(index) == '/') {
                if (index - start == 2
                        && name.charAt(start) == '.' && name.charAt(start + 1) == '.') {
                    throw new IllegalArgumentException("ZIP entry escapes its archive path: " + name);
                }
                start = index + 1;
            }
        }
    }
}

항목 이름은 상대 경로여야 하며 상위 디렉터리로 벗어나는 ..나 절대 경로는 거부합니다. 디렉터리 항목은 파일로 쓰지 않습니다.

같은 출력 경로는 실행기 하나에서 한 번만 쓸 수 있습니다. 두 ZIP에 data.csv가 있는데 ZIP 경로 접두사를 쓰지 않으면 Attempted to produce the same output twice가 날 수 있습니다. 다른 실행기에서 충돌하면 오류가 나지 않고 결과가 정의되지 않을 수 있습니다. 같은 ZIP 안에도 중복 논리 경로가 없어야 합니다.

모든 입력 파일을 ZIP 하나에 넣기

다음 myproject/files/ZipOnDriver.java는 입력의 각 파일을 하나의 bigzip.zip에 담습니다. 입력 파일이 ZIP이면 그 ZIP 바이트를 항목 하나로 보관하며 내부 항목들을 풀어 합치는 작업은 아닙니다.

package myproject.files;

import com.google.common.io.ByteStreams;
import com.palantir.transforms.lang.java.api.Compute;
import com.palantir.transforms.lang.java.api.FoundryInput;
import com.palantir.transforms.lang.java.api.FoundryOutput;
import com.palantir.transforms.lang.java.api.ReadOnlyLogicalFileSystem;
import com.palantir.transforms.lang.java.api.WriteOnlyLogicalFileSystem;
import com.palantir.util.syntacticpath.Paths;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;

public final class ZipOnDriver {
    @Compute
    public void compute(FoundryInput zipFiles, FoundryOutput output) {
        ReadOnlyLogicalFileSystem source = zipFiles.asFiles().getFileSystem();
        WriteOnlyLogicalFileSystem destination = output.getFileSystem();

        destination.writeTo(Paths.get("bigzip.zip"), stream -> {
            try (ZipOutputStream zip = new ZipOutputStream(stream)) {
                source.listAllFiles().forEach(path -> source.readFrom(path, input -> {
                    zip.putNextEntry(new ZipEntry(path.toString()));
                    ByteStreams.copy(input, zip);
                    zip.closeEntry();
                    return null;
                }));
            }
        });
    }
}

이 작업은 드라이버의 단일 스레드에서 수행합니다. 하나의 ZIP에 순서대로 기록하므로 병렬 입력 해제와 같은 처리량을 기대하지 않습니다. 목적이 압축 파일의 내용을 분석하는 것이라면 파일별 병렬 처리나 CSV 표 변환을 사용합니다.

파이프라인 등록

새 Java 하위 프로젝트에서 다음 myproject/FilePipelineDefiner.java를 사용할 수 있습니다. 가상 입출력 경로를 실제 자산으로 바꿉니다. 이미 PipelineDefiner가 있다면 구현을 둘 만들지 말고 기존 define()에 해당 등록을 합칩니다.

package myproject;

import com.palantir.transforms.lang.java.api.LowLevelTransform;
import com.palantir.transforms.lang.java.api.Pipeline;
import com.palantir.transforms.lang.java.api.PipelineDefiner;
import myproject.files.UnzipWithSpark;
import myproject.files.ZipOnDriver;

public final class FilePipelineDefiner implements PipelineDefiner {
    @Override
    public void define(Pipeline pipeline) {
        pipeline.register(LowLevelTransform.builder()
                .computeFunctionInstance(new UnzipWithSpark())
                .putParameterToInputAlias("zipFiles", "/Example Project/Inputs/archives")
                .putParameterToOutputAlias("output", "/Example Project/Outputs/unzipped")
                .build());
        pipeline.register(LowLevelTransform.builder()
                .computeFunctionInstance(new ZipOnDriver())
                .putParameterToInputAlias("zipFiles", "/Example Project/Inputs/archives")
                .putParameterToOutputAlias("output", "/Example Project/Outputs/combined_zip")
                .build());
    }
}

이 구현을 단독 진입점으로 쓴다면 ServiceLoader 파일의 한 줄도 myproject.FilePipelineDefiner로 지정합니다. 등록 계약의 서비스 파일 위치와 매개변수 이름 일치 조건을 따릅니다.