Java 증분 변환의 읽기와 쓰기

일반 변환은 입력 전체를 다시 계산하여 출력을 대체합니다. 증분 변환은 이전 빌드가 처리한 상태를 바탕으로 새 입력만 계산하거나 필요한 경우 전체 스냅샷을 다시 만듭니다.

Java는 Python의 증분 annotation처럼 런타임이 읽기 및 쓰기 방식을 자동 결정하는 모델을 사용하지 않습니다. 작성자가 입력 변경 유형을 보고 읽을 범위와 출력 모드를 함께 선택해야 합니다.

변경 유형과 읽기 범위

input.asDataFrame().modificationType()은 표 입력의 DataFrameModificationType을 제공합니다. 파일에는 대응하는 FilesModificationType이 있습니다.

readForRange()는 UNPROCESSED, PROCESSED, ENTIRE_VIEW를 받습니다. 각각 아직 처리하지 않은 부분, 이미 처리한 부분과 현재 전체 view를 나타냅니다.

출력 모드는 별도로 지정

쓰기 모드를 생략한 write()는 SNAPSHOT입니다. 새 입력만 읽고 기본 쓰기를 사용하면 기존 출력이 새 부분만으로 대체되므로 증분 처리가 되지 않습니다.

WriteMode.UPDATE와 DataFrameModificationType.UPDATED는 같은 개념이 아닙니다. 이 증분 쓰기의 UPDATE는 하위 변환에서 APPENDED로 보이는 추가 변경을 만들며, 입력의 UPDATED는 기존 데이터 수정도 포함할 수 있다는 신호입니다. 행별 SQL UPDATE와도 구분합니다.

전체 표 변환 예

다음은 갈색 눈인 행을 유지합니다. 가상 경로를 실제 자산으로 바꾸고 입력에 eye 열이 있어야 합니다. 파일은 myproject/datasets/IncrementalEyeFilter.java에 두고 자동 등록합니다.

package myproject.datasets;

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.Input;
import com.palantir.transforms.lang.java.api.Output;
import com.palantir.transforms.lang.java.api.ReadRange;
import com.palantir.transforms.lang.java.api.WriteMode;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public final class IncrementalEyeFilter {
    private static final Logger LOG = LoggerFactory.getLogger(IncrementalEyeFilter.class);

    @Compute
    public void compute(
            @Input("/Example Project/Inputs/students") FoundryInput input,
            @Output("/Example Project/Outputs/brown_eyes") FoundryOutput output) {
        ReadRange readRange;
        WriteMode writeMode;
        switch (input.asDataFrame().modificationType()) {
            case UNCHANGED:
            case APPENDED:
                readRange = ReadRange.UNPROCESSED;
                writeMode = WriteMode.UPDATE;
                break;
            case UPDATED:
            case NEW_VIEW:
                readRange = ReadRange.ENTIRE_VIEW;
                writeMode = WriteMode.SNAPSHOT;
                break;
            default:
                throw new IllegalArgumentException("Unknown input modification type");
        }
        LOG.info("Reading {} and writing {}", readRange, writeMode);
        Dataset<Row> rows = input.asDataFrame().readForRange(readRange);
        output.getDataFrameWriter(rows.filter("eye = 'Brown'")).write(writeMode);
    }
}

다음 가상 입력의 첫 빌드는 17, 18, 19를 포함하고, 두 번째 append에서는 새 갈색 눈 행인 20만 추가합니다. 기존 세 행은 유지됩니다.

파일을 증분 복사

파일 입력에서는 FoundryInput.asFiles()가 FoundryFiles를 반환합니다. 선택한 읽기 범위를 getFileSystem(readRange), 쓰기 모드를 출력의 getFileSystem(writeMode)에 전달합니다.

인자 없는 출력 getFileSystem()은 항상 SNAPSHOT입니다. 읽기만 UNPROCESSED로 바꾸어도 출력은 증분이 되지 않습니다.

package myproject.datasets;

import com.google.common.io.ByteStreams;
import com.palantir.transforms.lang.java.api.Compute;
import com.palantir.transforms.lang.java.api.FoundryFiles;
import com.palantir.transforms.lang.java.api.FoundryInput;
import com.palantir.transforms.lang.java.api.FoundryOutput;
import com.palantir.transforms.lang.java.api.Input;
import com.palantir.transforms.lang.java.api.Output;
import com.palantir.transforms.lang.java.api.ReadOnlyLogicalFileSystem;
import com.palantir.transforms.lang.java.api.ReadRange;
import com.palantir.transforms.lang.java.api.WriteMode;
import com.palantir.transforms.lang.java.api.WriteOnlyLogicalFileSystem;

public final class IncrementalFileCopyTransform {
    @Compute
    public void compute(
            @Input("/Example Project/Inputs/raw_files") FoundryInput input,
            @Output("/Example Project/Outputs/raw_files_copy") FoundryOutput output) {
        FoundryFiles files = input.asFiles();
        ReadRange readRange;
        WriteMode writeMode;
        switch (files.modificationType()) {
            case UNCHANGED:
            case APPENDED:
                readRange = ReadRange.UNPROCESSED;
                writeMode = WriteMode.UPDATE;
                break;
            case UPDATED:
            case NEW_VIEW:
                readRange = ReadRange.ENTIRE_VIEW;
                writeMode = WriteMode.SNAPSHOT;
                break;
            default:
                throw new IllegalArgumentException("Unknown file modification type");
        }
        ReadOnlyLogicalFileSystem source = files.getFileSystem(readRange);
        WriteOnlyLogicalFileSystem destination = output.getFileSystem(writeMode);
        source.listAllFiles().forEach(path -> source.readFrom(path, inputStream -> {
            destination.writeTo(
                    path, outputStream -> ByteStreams.copy(inputStream, outputStream));
            return null;
        }));
    }
}

이 코드는 필요한 파일만 읽고 원래 논리 경로로 씁니다. 기존 파일이 바뀌거나 새 view가 생기면 전체를 다시 복사합니다.

의도적인 스냅샷 전환

대부분 증분으로 실행하지만 가끔 전체를 다시 계산해야 한다면 별도의 trigger 데이터셋을 입력으로 둘 수 있습니다. 그 입력이 바뀌면 다른 입력들도 ENTIRE_VIEW로 읽고 SNAPSHOT으로 쓰도록 로직을 구성합니다. 읽기 범위와 쓰기 모드를 함께 바꿔야 합니다.

외부에서 provenance 없는 빈 append 트랜잭션을 만드는 방법으로 스냅샷을 강제할 수도 있지만 Java transforms API는 그 작업을 노출하지 않습니다.

고급 증분 옵션

다음은 저수준 변환에만 적용합니다. 자동 등록은 @Compute 메서드의 @UseIncrementalOptions(...), 수동 등록은 기존 LowLevelTransform.builder()의 옵션으로 설정합니다.

Annotation의 전체 형태는 @UseIncrementalOptions(IncrementalOptions.IGNORE_INCREMENTAL_DELETES) 또는 @UseIncrementalOptions(IncrementalOptions.USE_SCHEMA_MODIFICATION_TYPE)입니다. 두 유형은 com.palantir.transforms.lang.java.api에서 사용하며 기존 계산 로직과 입출력 등록은 그대로 필요합니다.

상위 삭제 무시

계속 커지는 append-only 입력의 오래된 데이터를 저장 공간 때문에 정리할 때 삭제가 증분 실행을 깨지 않도록 할 수 있습니다. 이 설정은 상위 삭제를 출력의 행 삭제로 동기화하는 기능이 아닙니다. 이미 계산한 결과를 유지하는 의도와 맞는 경우에만 사용합니다.

스키마 변경 분리

보통 입력 스키마 변경은 DataFrameModificationType.NEW_VIEW를 만들 수 있습니다. USE_SCHEMA_MODIFICATION_TYPE을 켜면 스키마 변경을 UNCHANGED와 SchemaModificationType.NEW_SCHEMA 신호로 구분하여 처리할 수 있습니다.

변환이 바뀐 열을 직접 사용하거나 그 열이 출력에 포함되어 있으면 여전히 실패할 수 있습니다. 예를 들어 eye를 삭제하거나 이름을 바꾸면 갈색 눈 필터가 동작하지 않습니다. hair를 그대로 출력하던 로직에서 hair가 없어져도 출력 스키마에 영향이 있습니다.

반면 최초 스냅샷부터 id, eye만 선택한 변환은 사용하지 않는 hair 열의 제거에 영향을 받지 않을 수 있습니다. 기존 요구와 충돌하지 않는 열 추가도 증분을 유지할 수 있는 사례입니다. 관련 열의 파괴적 변경 뒤에는 새 스냅샷이 필요합니다.

고급 옵션은 데이터 삭제와 스키마 변화에 대한 자동 안전 경계를 완화합니다. 실제 의미를 확신할 수 없는 경우에는 옵션으로 실패를 숨기기보다 전체 재계산과 지원 검토를 선택합니다.