DuckDB로 Foundry 데이터 처리하기

DuckDB 통합은 입력을 지연 다운로드하고 스트리밍 방식으로 읽는 단일 노드 계산 경로입니다. 큰 입력을 SQL로 처리하거나 엔진의 메모리와 병렬성을 조정해야 할 때 검토합니다. 실제 성능은 쿼리와 자료형에 따라 Pandas, Polars와 비교합니다.

@transform.using()은 기본적으로 lightweight 변환을 선언합니다. 전달되는 ctx는 LightweightContext이며 출력은 LightweightOutput입니다. ctx.duckdb().conn에서 얻은 연결에는 모든 입력 데이터셋이 아직 구체화하지 않은 테이블로 등록되고, 테이블 이름은 decorator에서 정한 입력 인자 이름을 따릅니다.

등록된 입력을 SQL로 읽기

다음 예제의 자산 경로는 실제 프로젝트 경로로 바꿉니다.

from transforms.api import Input, Output, transform


@transform.using(
    source=Input("/Example Project/Inputs/source"),
    output=Output("/Example Project/Outputs/duckdb_copy"),
)
def compute(ctx, source, output):
    connection = ctx.duckdb().conn
    result = connection.sql("SELECT * FROM source")
    output.write_table(result)

직접 다운로드하거나 입력마다 테이블을 등록할 필요가 없습니다. DuckDB 자체의 Python relational API도 사용할 수 있습니다. Ibis와 SQLFrame은 DataFrame 표현식이나 PySpark와 유사한 문법을 사용하려는 경우의 추가 선택지입니다.

증분 읽기 모드

증분 실행의 기본 읽기는 added, 비증분 실행의 기본 읽기는 current입니다. 특정 입력을 항상 전체 상태로 읽으려면 read_modes의 키에 입력 핸들을, 값에 읽기 모드를 지정합니다.

from transforms.api import Input, Output, incremental, transform


@incremental()
@transform.using(
    source=Input("/Example Project/Inputs/source"),
    output=Output("/Example Project/Outputs/current_snapshot"),
)
def compute(ctx, source, output):
    connection = ctx.duckdb(read_modes={source: "current"}).conn
    result = connection.sql("SELECT * FROM source")
    output.set_mode("replace")
    output.write_table(result)

입력을 전체로 읽고 그 결과 전체를 출력한다면 기존 출력에 계속 덧붙이지 않도록 쓰기 모드도 맞춥니다. 일부 입력을 참조 테이블로 전체 읽고 다른 입력의 새 행만 계산하는 경우에는 계산 결과의 의미에 따라 출력 모드를 정합니다.

SQL 파일로 분리하기

sql_from_file()은 호출한 Python 파일을 기준으로 상대 경로를 해석합니다. 아래 두 파일을 나란히 둡니다. 입력에는 숫자 열 column_a가 있어야 합니다.

query.sql:

SELECT * FROM source WHERE column_a > 100;

Python 변환:

from transforms.api import Input, Output, incremental, transform


@incremental()
@transform.using(
    source=Input("/Example Project/Inputs/source"),
    output=Output("/Example Project/Outputs/large_values"),
)
def compute(ctx, source, output):
    result = ctx.duckdb().conn.sql_from_file("query.sql")
    output.write_table(result)

SQL 파일을 배포 대상에 포함하고, 파일 변경에 따라 변환이 다시 계산되어야 하면 외부 파일 버전 추적의 tllvFiles도 설정합니다.

메모리와 실행 스레드

연결을 얻은 뒤 설정 SQL을 실행합니다. 다음 두 문장은 변환 함수 안에서 얻은 connection에 적용하는 예시입니다.

connection.execute("SET memory_limit = '2GB'")
connection.execute("SET threads = 4")

메모리를 줄이면 디스크 사용과 처리 시간이 증가할 수 있습니다. memory_limit은 주로 버퍼 관리자의 한도이며 Python 객체, 결과와 일부 집계 상태를 포함한 프로세스 전체 메모리의 절대 상한은 아닙니다. 요청한 컨테이너 메모리에 여유를 두고 스레드 수를 함께 조정합니다.

원시 파일 읽기와 COPY 출력

입력 이름이 source이면 사전 구성된 source_files() 함수가 실제 데이터 파일 경로 목록을 반환합니다. 이를 DuckDB의 read_parquet()나 read_csv()에 넘겨 파일 읽기 옵션을 직접 지정할 수 있습니다.

출력 경로는 두 가지입니다.

다음 두 예제는 별도의 출력 방식입니다. 한 변환에서 같은 결과를 두 번 기록하는 절차로 함께 실행하지 않습니다. 입력에는 스키마를 읽을 수 있는 Parquet 파일이 하나 이상 있어야 하며 my_column 문자열 열이 필요합니다.

객체 저장소 출력 경로는 현재 출력 트랜잭션에 연결된 로컬 S3 프록시의 가상 경로입니다. 매 실행에서 출력 핸들이 제공하는 값을 사용합니다.

로컬 디스크를 거쳐 출력

from transforms.api import Input, Output, transform


@transform.using(
    source=Input("/Example Project/Inputs/source"),
    output=Output("/Example Project/Outputs/local_copy"),
)
def compute(ctx, source, output):
    connection = ctx.duckdb().conn
    connection.execute(
        """
        COPY (
            SELECT *
            FROM read_parquet(source_files(), file_row_number = true)
            WHERE my_column = 'abc'
            LIMIT 10
        ) TO ? (FORMAT PARQUET, PER_THREAD_OUTPUT true)
        """,
        [output.path_for_write_table],
    )
    output.write_table(output.path_for_write_table)

file_row_number=true는 파일 내 행 번호를 함께 읽습니다. PER_THREAD_OUTPUT은 쓰기 스레드마다 파일을 생성합니다. 정렬하지 않은 LIMIT 10은 특정 행 순서를 보장하지 않습니다.

객체 저장소로 직접 출력

이 예제에는 추가로 date 분할 열이 필요합니다.

from transforms.api import Input, Output, incremental, transform


@incremental()
@transform.using(
    source=Input("/Example Project/Inputs/source"),
    output=Output("/Example Project/Outputs/partitioned_copy"),
)
def compute(ctx, source, output):
    connection = ctx.duckdb().conn
    connection.execute(
        """
        COPY (
            SELECT *
            FROM read_parquet(source_files(), file_row_number = true)
            WHERE my_column = 'abc'
            LIMIT 10
        ) TO ? (
            FORMAT PARQUET,
            PARTITION_BY (date),
            WRITE_PARTITION_COLUMNS true,
            FILENAME_PATTERN 'file_{uuid}'
        )
        """,
        [output.path_for_object_store_write_table],
    )
    copied_rows = connection.fetchone()[0]
    if copied_rows == 0:
        empty = connection.sql(
            "SELECT * FROM read_parquet(source_files(), file_row_number = true) LIMIT 0"
        )
        output.write_table(empty)
    else:
        output.put_metadata()

파일을 업로드했으면 put_metadata()로 스키마를 추론합니다. 필터 결과가 비어 있으면 분할 COPY가 파일을 만들지 않을 수 있으므로, 같은 열 스키마의 빈 relation을 write_table()로 기록합니다. put_metadata()는 성공적으로 업로드한 Parquet 파일이 하나 이상 있어야 사용할 수 있습니다.

파일명에는 DuckDB의 UUID 패턴을 사용하여 앞선 증분 트랜잭션의 파일과 충돌하거나 덮어쓰지 않게 합니다. 직접 지정한 고정 파일명을 증분 실행마다 반복해서 쓰면 충돌 위험이 있습니다.

분할 쓰기에서는 WRITE_PARTITION_COLUMNS=true를 권장합니다. 분할 열을 파일에도 저장하여 다른 변환과의 호환성을 높입니다. NULL 분할 키는 경로 표기를 별도로 확인합니다. null로 생성된 경로와 Hive의 __HIVE_DEFAULT_PARTITION__ 표준 표기를 하위 읽기 엔진이 동일하게 해석한다고 가정하지 않습니다. 업무상 허용되는 경우 분할 열의 NULL을 명시적인 값으로 채울 수 있지만 원래 NULL과 일반 값을 구분할 수 있어야 합니다.

먼저 내려받고 독립 연결에서 읽기

전체 입력을 시작 시점에 디스크로 내려받으려면 source.path()를 호출합니다. 다음 예제는 내려받은 Parquet에서 Name이 John으로 시작하는 행을 선택합니다.

from pathlib import Path

import duckdb
from transforms.api import Input, Output, transform


@transform.using(
    source=Input("/Example Project/Inputs/people"),
    output=Output("/Example Project/Outputs/eager_john_names"),
)
def compute(source, output):
    parquet_glob = str(Path(source.path()) / "**" / "*.parquet")
    with duckdb.connect(database=":memory:") as connection:
        connection.execute(
            """
            COPY (
                SELECT * FROM parquet_scan($source_files) WHERE Name LIKE $name_prefix
            ) TO $output_path (FORMAT PARQUET, PER_THREAD_OUTPUT true)
            """,
            {
                "source_files": parquet_glob,
                "name_prefix": "John%",
                "output_path": output.path_for_write_table,
            },
        )
    output.write_table(output.path_for_write_table)

입력 경로와 출력 경로는 SQL 매개변수로 전달합니다. 이 연결은 예제에서 직접 생성하므로 작업 후 닫습니다. 기본 lazy 통합과 달리 전체 입력을 내려받을 디스크 공간이 필요합니다.