저장소 변환의 CSV, 여러 출력과 압축 파일

Workbook에서 시험한 로직을 Code Repository로 옮기면 출력 형식, 여러 출력과 파일 단위 처리를 명시적으로 구성할 수 있습니다. 다음 예제는 별도 표시가 없는 한 Code Repositories의 변환 코드입니다. Workbook 노드 하나가 여러 DataFrame을 반환할 수 있다는 뜻은 아닙니다.

가상 프로젝트 경로는 실제 입력 및 출력으로 바꾸고 해당 언어의 변환을 파이프라인에 등록해야 합니다.

CSV 출력

Python

from transforms.api import Input, Output, transform


@transform(
    output=Output("/Example Project/Outputs/python_csv"),
    source=Input("/Example Project/Inputs/table"),
)
def write_csv(output, source):
    output.write_dataframe(source.dataframe(), output_format="csv")

Java

DatasetFormatSettings에 format("csv")를 지정합니다. 다음 클래스는 자동 등록용이며 Java 소스 폴더에 둡니다.

package myproject.datasets;

import com.palantir.foundry.spark.api.DatasetFormatSettings;
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;

public final class CsvOutput {
    @Compute
    public void compute(
            @Input("/Example Project/Inputs/table") FoundryInput input,
            @Output("/Example Project/Outputs/java_csv") FoundryOutput output) {
        output.getDataFrameWriter(input.asDataFrame().read())
                .setFormatSettings(DatasetFormatSettings.builder().format("csv").build())
                .write();
    }
}

SQL

CREATE TABLE `/Example Project/Outputs/sql_csv`
USING CSV
AS SELECT * FROM `/Example Project/Inputs/table`

CSV 형식 선택과 파일 개수는 별개입니다. 파티션 하나로 줄여 내려받을 파일을 만들려면 다운로드 파일 준비의 비용과 옵션을 적용합니다.

여러 변환 생성

입출력 조합마다 별도의 변환이 필요하면 generator로 정의들을 만들 수 있습니다. 다음은 src1, src2, src3의 입력에서 source 값이 같은 행만 각 출력으로 쓰는 예입니다. 파일 위치는 myproject/datasets/source_filters.py입니다.

from pyspark.sql import functions as F
from transforms.api import Input, Output, transform_df


def create_source_transform(source_name):
    @transform_df(
        Output(f"/Example Project/Outputs/{source_name}"),
        my_input=Input(f"/Example Project/Inputs/{source_name}"),
    )
    def compute(my_input):
        return my_input.filter(F.col("source") == source_name)

    compute.__name__ = f"filter_{source_name}"
    return compute


TRANSFORMS = [create_source_transform(name) for name in ["src1", "src2", "src3"]]

각 factory 호출이 source_name을 따로 보관하므로 반복문의 마지막 이름으로 모든 변환이 바뀌지 않습니다.

파이프라인 진입점에서 명시적으로 등록하려면 다음과 같이 구성합니다. 기존 진입점이 있다면 Pipeline 객체를 중복 생성하지 말고 기존 객체에 추가합니다.

from transforms.api import Pipeline
from myproject.datasets import source_filters

my_pipeline = Pipeline()
my_pipeline.add_transforms(*source_filters.TRANSFORMS)

저장소의 entry point는 이 모듈의 실제 Pipeline 이름을 가리켜야 합니다. 등록된 변환 하나마다 별도 출력과 빌드 단위를 가집니다.

한 변환에서 출력 둘 만들기

같은 입력을 같은 job에서 처리하여 여러 출력을 쓰려면 @transform에 출력 핸들을 여러 개 선언합니다. 다음 입력에는 hair, sex 열이 있고 값은 예시 분류인 Brown, Male, Female을 사용한다고 가정합니다.

from pyspark.sql import functions as F
from transforms.api import Input, Output, transform


@transform(
    students=Input("/Example Project/Inputs/students_hair_eye_color"),
    males=Output("/Example Project/Outputs/brown_hair_males"),
    females=Output("/Example Project/Outputs/brown_hair_females"),
)
def brown_hair_by_sex(students, males, females):
    rows = students.dataframe()
    brown_hair = rows.filter(F.col("hair") == "Brown")
    males.write_dataframe(brown_hair.filter(F.col("sex") == "Male"))
    females.write_dataframe(brown_hair.filter(F.col("sex") == "Female"))

가상 입력의 앞 세 행이 다음과 같다면 갈색 머리인 2번 행만 이 부분의 필터를 통과합니다. 실제 데이터 전체의 결과는 모든 입력 행으로 계산합니다.

다중 출력 설계에서 공통 계산 재사용, 실행 주기와 실패 결합을 고려합니다. 같은 DataFrame 변수를 사용했다는 것만으로 Spark 계산이 자동 캐시되는 것은 아닙니다.

GZIP을 스트림으로 읽기

Foundry 파일 입력은 스트림으로 읽을 수 있습니다. gzip.GzipFile과 io.BufferedReader는 압축을 풀면서 바이트를 제공하며, 텍스트가 필요하면 io.TextIOWrapper로 감쌉니다. 파일 전체를 한꺼번에 메모리에 읽거나 임시 디스크에 복사할 필요는 없습니다.

다음은 UTF-8 GZIP 텍스트를 파일 경로와 줄 문자열의 표로 만드는 전체 변환입니다. 입력 데이터셋에는 GZIP 파일들만 있어야 합니다.

import gzip
import io

from transforms.api import Input, Output, transform


@transform(
    output=Output("/Example Project/Outputs/gzip_lines"),
    source=Input("/Example Project/Inputs/gzip_text"),
)
def read_gzip_lines(ctx, output, source):
    filesystem = source.filesystem()
    encoding = "utf-8"

    def process_file(file_status):
        with filesystem.open(file_status.path, "rb") as raw:
            with gzip.GzipFile(fileobj=raw, mode="rb") as decompressed:
                with io.BufferedReader(decompressed) as buffered:
                    with io.TextIOWrapper(buffered, encoding=encoding) as text:
                        for line in text:
                            yield (file_status.path, line)

    rows = filesystem.files().rdd.flatMap(process_file)
    frame = ctx.spark_session.createDataFrame(
        rows, schema="file_path string, line string"
    )
    output.write_dataframe(frame)

EBCDIC CP500으로 저장된 텍스트라면 encoding을 CP500으로 지정합니다. 바이너리 결과를 처리할 때는 TextIOWrapper 없이 BufferedReader에서 바이트를 읽습니다. GzipFile을 닫는 것만으로 원래 fileobj가 닫히지 않으므로 원천 파일도 별도 context manager로 닫습니다.

ZIP의 병렬 처리 범위

Java의 filesAsDataset()은 ZIP 파일별로 작업을 나눌 수 있습니다. ZipInputStream으로 각 항목을 읽고 원래 ZIP의 논리 경로를 접두사로 사용하여 출력 충돌을 막습니다. 전체 Java 해제 구현에 파이프라인 등록과 경로 조건을 포함했습니다.

ZIP 한 개의 압축 내부를 여러 task로 분할하는 방식은 아닙니다. 한 압축 파일 안의 분산 처리가 필요하면 reader가 지원하는 splittable 형식, 예를 들어 bzip2를 검토합니다. 압축한 CSV를 표로 바로 바꾸는 경로는 Java CSV reader를 사용합니다.