파일 수를 기준으로 증분 출력 병합하기

증분 출력을 계속 추가하면 작은 파일이 쌓입니다. 이전 출력의 파일 수가 기준을 넘었을 때 이전 행과 새 행을 합쳐 replace로 다시 쓰고, 그 외에는 새 행만 modify로 추가할 수 있습니다.

다음 네 예제는 입력 행을 그대로 보존하면서 출력 파일을 정리하는 변환입니다. 입력은 새 행이 추가되는 표이며, 스냅샷 원천을 매번 전체로 읽는 이력 누적과는 입력 모드가 다릅니다. 기준은 파일 수가 100개를 초과하는 경우입니다. 첫 실행과 비증분 실행에서는 이전 출력에 덧붙이지 않고 전체 입력으로 교체합니다.

스키마와 경량 경로의 비용

입력과 이전 출력의 열은 호환되어야 합니다. 예시 논리 열과 대표적인 엔진 표현은 다음과 같습니다.

Lightweight 출력의 polars()와 pandas()는 읽기 모드를 키워드로 지정합니다. 빈 출력용 schema는 위 엔진별 사전이 아니라 FoundryFieldSchema입니다. 아래 코드는 실제 증분 실행에서만 기존 출력을 읽으므로 별도의 빈 출력 schema를 만들지 않습니다.

Lightweight의 filesystem()에는 mode 인자가 없습니다. 다음 공통 함수는 v2_semantics=True에서 지원하는 path(mode="previous")로 이전 파일을 내려받아 개수를 셉니다. 따라서 병합하지 않는 실행에서도 이전 파일의 전체 다운로드 비용이 먼저 듭니다. 메타데이터만 조회하여 파일 수를 판단하는 방법과 같은 비용으로 해석하지 않습니다.

src/myproject/datasets/compaction_helpers.py:

from pathlib import Path

FILE_COUNT_LIMIT = 100


def should_compact(output):
    previous_directory = Path(output.path(mode="previous"))
    file_count = sum(
        1 for path in previous_directory.rglob("*") if path.is_file()
    )
    return file_count > FILE_COUNT_LIMIT

이 공통 함수를 사용하는 경량 예제는 모두 v2를 명시합니다. 기존 증분 변환에 v2를 새로 켜면 다음 빌드가 한 번 스냅샷으로 실행되므로 기존 이력을 복원할 수 있는 입력인지 확인합니다.

Polars

import polars as pl
from transforms.api import Input, Output, incremental, transform
from myproject.datasets.compaction_helpers import should_compact


@incremental(v2_semantics=True, semantic_version=1)
@transform.using(
    source=Input("/Example Project/Inputs/measurements"),
    output=Output("/Example Project/Outputs/compact_polars"),
)
def compute(ctx, source, output):
    rows = source.polars()
    if not ctx.is_incremental:
        mode = "replace"
    elif should_compact(output):
        rows = pl.concat([rows, output.polars(mode="previous")])
        mode = "replace"
    else:
        mode = "modify"
    output.set_mode(mode)
    output.write_table(rows)

DuckDB와 Polars 결과

DuckDB에서 새 입력을 SQL로 읽고 Polars DataFrame으로 수집하여 이전 출력과 합칩니다. 수집 결과와 병합 대상이 단일 노드 메모리에 들어가야 합니다.

import polars as pl
from transforms.api import Input, Output, incremental, transform
from myproject.datasets.compaction_helpers import should_compact


@incremental(v2_semantics=True, semantic_version=1)
@transform.using(
    source=Input("/Example Project/Inputs/measurements"),
    output=Output("/Example Project/Outputs/compact_duckdb"),
)
def compute(ctx, source, output):
    rows = ctx.duckdb().conn.sql("SELECT * FROM source").pl()
    if not ctx.is_incremental:
        mode = "replace"
    elif should_compact(output):
        rows = pl.concat([rows, output.polars(mode="previous")])
        mode = "replace"
    else:
        mode = "modify"
    output.set_mode(mode)
    output.write_table(rows)

Pandas

Pandas의 concat(..., ignore_index=True)로 행을 합칩니다. 이전 출력의 index 값을 데이터의 고유 식별자로 취급하지 않습니다.

import pandas as pd
from transforms.api import Input, Output, incremental, transform
from myproject.datasets.compaction_helpers import should_compact


@incremental(v2_semantics=True)
@transform.using(
    source=Input("/Example Project/Inputs/measurements"),
    output=Output("/Example Project/Outputs/compact_pandas"),
)
def compute(ctx, source, output):
    rows = source.pandas()
    if not ctx.is_incremental:
        mode = "replace"
    elif should_compact(output):
        rows = pd.concat(
            [rows, output.pandas(mode="previous")], ignore_index=True
        )
        mode = "replace"
    else:
        mode = "modify"
    output.set_mode(mode)
    output.write_table(rows)

PySpark

Spark의 증분 출력은 filesystem(mode="previous")로 이전 파일 목록을 읽을 수 있습니다. 병합 시 사용할 이전 출력 schema는 실제 열 자료형과 일치해야 하며 모든 열을 nullable로 선언합니다.

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

FILE_COUNT_LIMIT = 100
SCHEMA = T.StructType([
    T.StructField("Value", T.DoubleType(), True),
    T.StructField("Time", T.TimestampType(), True),
    T.StructField("DataGroup", T.StringType(), True),
])


@incremental()
@transform.spark.using(
    source=Input("/Example Project/Inputs/measurements"),
    output=Output("/Example Project/Outputs/compact_spark"),
)
def compute(ctx, source, output):
    rows = source.dataframe()
    if not ctx.is_incremental:
        mode = "replace"
    else:
        previous_files = output.filesystem(mode="previous").ls()
        file_count = sum(1 for _ in previous_files)
        if file_count > FILE_COUNT_LIMIT:
            previous = output.dataframe("previous", schema=SCHEMA)
            rows = rows.unionByName(previous)
            mode = "replace"
        else:
            mode = "modify"
    output.set_mode(mode)
    output.write_dataframe(rows.select(SCHEMA.fieldNames()).coalesce(1))

이 Spark 예제는 출력 파티션을 하나로 줄입니다. 전체 이력이 커지면 한 파티션의 처리량과 메모리가 제약이 되므로, 실제 크기에 맞는 파티션 수를 선택합니다. 파일 수는 writer의 파일 크기/행 수 설정에도 영향을 받습니다. 경량 엔진의 replace 역시 파일을 다시 쓰는 방식이며 결과 파일 수는 해당 writer 설정에 따릅니다.