증분 추가, 중복 제거와 집계

증분 계산 방식은 새 결과가 무엇에 의존하는지에 따라 달라집니다. 새 행만으로 계산할 수 있으면 추가하고, 기존 출력과 비교해야 하면 이전 결과를 읽습니다. 집계처럼 기존 행의 값도 바뀌면 계산한 전체 상태로 출력을 교체합니다.

읽기 모드의 예시 데이터

다음은 가상의 학생 자료입니다. 첫 스냅샷은 1~3번이고 추가 배치는 4~6번입니다.

id	hair	eye	sex
1	Brown	Green	Female
2	Red	Blue	Male
3	Blond	Hazel	Female
4	Brown	Green	Female
5	Brown	Blue	Male
6	Blond	Hazel	Female

실행 상태	`previous` 입력	`current` 입력	`added` 입력
첫 스냅샷 처리	빈 입력	1~3번	1~3번
첫 배치를 처리한 뒤 추가 배치 처리	1~3번	1~6번	4~6번

모드를 생략하면 added와 같은 범위를 읽습니다. 같은 Python 객체 인스턴스를 반환하는지에 의존할 필요는 없습니다.

새 입력만 처리하여 추가하기

행별 자료형 변환, 날짜 문자열 형식화와 필터는 새 입력에서 새 결과를 계산할 수 있습니다. 네 엔진의 갈색 머리 필터는 증분 실행에서 새 입력을 필터하여 추가하고, 비증분 실행에서는 전체 입력을 필터하여 교체합니다. 첫 결과는 1번이며 다음 추가 결과는 4번과 5번입니다.

기존 출력과 비교하여 중복 제거하기

새 배치 안의 중복을 먼저 없애고 이전 출력에 이미 있는 행을 뺀 뒤 추가합니다. 이전 출력 자체가 올바르게 중복 제거되어 있다는 전제입니다. 전체 행을 기준으로 하므로 id만 같은 서로 다른 행을 같은 레코드로 취급하지 않습니다.

from pyspark.sql import types as T
from transforms.api import Input, Output, incremental, transform


@incremental()
@transform.spark.using(
    students=Input("/Example Project/Inputs/students"),
    processed=Output("/Example Project/Outputs/distinct_students"),
)
def compute(students, processed):
    added = students.dataframe()
    previous_schema = T.StructType([
        T.StructField(field.name, field.dataType, True, field.metadata)
        for field in added.schema.fields
    ])
    previous = processed.dataframe("previous", schema=previous_schema)
    processed.write_dataframe(added.distinct().subtract(previous))

첫 실행처럼 이전 결과가 없을 때도 빈 DataFrame을 만들 수 있도록 schema를 제공합니다. 저장된 출력은 nullable이므로 이전 읽기 schema의 열도 nullable로 맞춥니다.

새 집계를 이전 집계와 합쳐 교체하기

머리색별 인원수는 새 배치의 인원수를 계산하고 이전 인원수와 합산합니다. 전체 학생을 다시 읽지는 않지만 출력 집계는 전체 상태를 교체합니다.

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

COUNTS_SCHEMA = T.StructType([
    T.StructField("hair", T.StringType(), True),
    T.StructField("count", T.LongType(), True),
])


@incremental()
@transform.spark.using(
    students=Input("/Example Project/Inputs/students"),
    processed=Output("/Example Project/Outputs/hair_counts"),
)
def compute(students, processed):
    added_counts = students.dataframe().groupBy("hair").count()
    previous = processed.dataframe("previous", schema=COUNTS_SCHEMA)
    totals = (
        added_counts.unionByName(previous)
        .groupBy("hair")
        .agg(F.sum("count").alias("count"))
    )
    processed.set_mode("replace")
    processed.write_dataframe(totals.select(COUNTS_SCHEMA.fieldNames()))

최종 출력은 Brown 3명, Red 1명, Blond 2명입니다. 같은 집계 행을 modify로 계속 추가하면 현재 합계가 여러 버전으로 쌓이므로 replace를 명시합니다.

서로 다른 CSV 열을 증분으로 수용하기

스키마 없는 입력 파일을 표로 변환할 때는 고정 schema에 맞춰 추가 열을 무시하고 누락 열을 채우거나, 들어온 열을 동적으로 수용할 수 있습니다. 동적 수용에서는 다음 조건을 만족해야 합니다.

기존 출력과 이름이 같은 열은 자료형도 같아야 합니다.
기존 출력의 모든 열이 새 배치에도 있어야 합니다. 누락 열은 같은 자료형의 NULL로 채울 수 있습니다.
기존 value와 새 Value처럼 대소문자만 다른 열을 만들지 않습니다.

다음 구현은 헤더가 있는 CSV를 모두 문자열로 읽어 파일마다 자료형 추론이 달라지는 문제를 피합니다. Parquet에 맞게 열 이름을 정리한 뒤 소문자로 통일하며, 이번 배치의 파일은 누락 열을 허용하여 합칩니다. 한 파일 안에 대소문자만 다른 열이 이미 함께 들어 있는 경우는 원천 헤더를 먼저 정리해야 합니다.

from functools import reduce

from pyspark.sql import functions as F
from transforms.api import Input, Output, incremental, transform
from transforms.verbs.dataframes import sanitize_schema_for_parquet


def quoted_column(name):
    return F.col("`" + name.replace("`", "``") + "`")


@incremental()
@transform.spark.using(
    csvs=Input("/Example Project/Inputs/csv_files"),
    parsed=Output("/Example Project/Outputs/parsed_csvs"),
)
def compute(ctx, csvs, parsed):
    filesystem = csvs.filesystem()

    def read_file(path):
        frame = (
            ctx.spark_session.read
            .option("inferSchema", False)
            .option("header", True)
            .csv(filesystem.hadoop_path + "/" + path)
        )
        cleaned = sanitize_schema_for_parquet(frame)
        return cleaned.select(*[
            quoted_column(name).alias(name.lower()) for name in cleaned.columns
        ])

    frames = [read_file(item.path) for item in filesystem.ls()]
    if not frames:
        ctx.abort_job()
        return

    batch = reduce(
        lambda left, right: left.unionByName(right, allowMissingColumns=True),
        frames,
    )
    if ctx.is_incremental:
        existing_columns = parsed.dataframe("previous").columns
        new_columns = set(batch.columns)
        batch = batch.select(
            *[quoted_column(name) for name in batch.columns],
            *[
                F.lit(None).cast("string").alias(name)
                for name in existing_columns if name not in new_columns
            ],
        )
    parsed.write_dataframe(batch)

이전 schema를 지정하지 않고 previous를 읽는 부분은 실제 증분 실행에서만 수행합니다. 첫 실행에서는 이번 CSV들의 schema로 출력을 만듭니다. 입력 파일이 없으면 작업 전체를 중단하여 이전 출력을 보존합니다.

열과 계산 로직 변경

호환되는 새 출력 열을 추가해도 증분 여부 자체가 무효화되지는 않습니다. 새 배치에는 값이 생기지만 이미 기록한 행의 새 열은 NULL입니다. 과거 행까지 채우려면 semantic version을 변경하여 한 번 전체 입력으로 다시 계산합니다.

스냅샷 입력을 쌓아 이력 데이터셋을 만든 경우에는 과거 스냅샷을 현재 입력에서 모두 복원할 수 없습니다. 단순한 전체 재계산으로 과거 이력이 복구된다고 가정하지 말고 데이터 소유자와 Palantir 담당자에게 이력 보존과 변경 방법을 확인합니다.

양쪽 입력에 지연 도착이 있는 조인은 별도 처리 조건이 필요합니다. 브랜치에서 이전 상태를 이어 쓰는 동작과 트랜잭션 오류는 증분 오류에서 설명합니다.

Keyboard shortcuts