Polars의 지연 실행과 스트리밍

Polars의 lazy API는 연산을 실행 계획으로 모아 두었다가 결과를 수집하거나 출력할 때 계산합니다. 엔진은 계획 전체를 보고 필터와 열 선택을 읽기 단계 가까이 옮길 수 있습니다. 큰 데이터나 연산이 여러 단계로 이어지는 파이프라인에 권장되지만, 쿼리에 따라 효과가 다르므로 운영에 반영하기 전에 해당 파이프라인의 성능과 안정성을 확인합니다.

Foundry 입력을 지연 읽기로 열기

input.polars(lazy=True)는 LazyFrame을 반환합니다. 기본값은 lazy=False입니다. 아래 가상 경로를 실제 자산으로 바꾸면 입력 전체를 지연 읽기로 열어 출력 데이터셋에 복사합니다.

from transforms.api import Input, Output, transform


@transform.using(
    source=Input("/Example Project/Inputs/source"),
    output=Output("/Example Project/Outputs/copy"),
)
def compute(source, output):
    query = source.polars(lazy=True)
    output.write_table(query)

lazy 입력은 내부 객체 저장소 프록시를 통해 필요한 데이터만 읽을 수 있게 노출됩니다. eager 모드는 입력 전체를 로컬 디스크로 미리 내려받습니다. Foundry의 Polars lazy 입출력은 스트리밍을 사용하도록 구성되므로 데이터를 여러 배치로 처리할 수 있습니다. 다만 결과를 직접 collect()하면 결과 전체가 메모리에 구체화되며, 각 연산의 지원 여부와 메모리 요구량도 확인해야 합니다.

필터를 읽기 단계로 내리기

Predicate pushdown은 .filter(), .where()와 불리언 조건을 계획에 기록한 뒤 가능한 한 일찍 적용하는 최적화입니다. 필터를 파일 스캔까지 내릴 수 있으면 읽지 않아도 되는 데이터의 입출력을 줄입니다. 전체 중 작은 일부만 사용하는 쿼리일수록 효과가 커질 수 있습니다. Projection pushdown은 필요한 열만 읽도록 줄이는 최적화입니다.

다음 예제는 date 열이 Polars Date인 판매 데이터에서 2025년 6월 1일부터 30일까지를 선택합니다. 날짜형 열에는 같은 자료형의 경계값을 사용합니다.

from datetime import date

import polars as pl
from transforms.api import Input, Output, transform


@transform.using(
    sales=Input("/Example Project/Inputs/sales"),
    output=Output("/Example Project/Outputs/june_2025_sales"),
)
def compute(sales, output):
    query = sales.polars(lazy=True).filter(
        (pl.col("date") >= pl.lit(date(2025, 6, 1)))
        & (pl.col("date") <= pl.lit(date(2025, 6, 30)))
    )
    output.write_table(query)

문자열 열을 비교하려면 값과 경계가 모두 YYYY-MM-DD처럼 동일하게 정규화된 형식이어야 합니다. 시간까지 포함한 Timestamp 열에서는 날짜 경계가 포함하는 시간 범위도 별도로 정합니다.

실행 계획과 중간 결과 보기

LazyFrame.explain()은 계획과 적용한 최적화를 보여 줍니다. 다음 코드는 계획을 출력한 뒤 같은 쿼리를 출력 데이터셋에 씁니다.

import polars as pl
from transforms.api import Input, Output, transform


@transform.using(
    source=Input("/Example Project/Inputs/sales"),
    output=Output("/Example Project/Outputs/positive_sales"),
)
def compute(source, output):
    query = source.polars(lazy=True).filter(pl.col("amount") > 0)
    print(query.explain())
    output.write_table(query)

문제가 생기는 단계를 나누어 보려면 중간 결과를 collect()하여 확인합니다. amount가 숫자이고 date 열이 있는 입력을 사용하는 다음 예제에서는 필터 결과를 한 번 구체화한 뒤 그 결과에서 두 열을 선택합니다.

import polars as pl
from transforms.api import Input, Output, transform


@transform.using(
    source=Input("/Example Project/Inputs/sales"),
    output=Output("/Example Project/Outputs/positive_sales_debug"),
)
def compute(source, output):
    intermediate = (
        source.polars(lazy=True)
        .filter(pl.col("amount") > 0)
        .collect()
    )
    print(intermediate.shape)
    output.write_table(intermediate.lazy().select("date", "amount"))

이 구체화 지점은 메모리를 사용하고 앞뒤 연산을 한 계획으로 최적화할 기회도 줄입니다. 오류 조사나 작은 입력의 단계별 탐색에는 eager 모드가 더 단순할 수 있습니다.

스트리밍 데이터셋의 빈 증분 구간

증분 lightweight 변환에서 새 파일이 없는 스트리밍 구간에 polars(lazy=True)를 호출하면 BinderException이 발생할 수 있습니다. 증분 실행인지 확인하고 먼저 eager 입력이 비었는지 검사합니다. 아래 예제는 새 데이터가 없으면 작업 전체의 트랜잭션을 중단하여 기존 출력을 유지합니다.

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


@incremental()
@transform.using(
    source=Input("/Example Project/Inputs/stream_archive"),
    output=Output("/Example Project/Outputs/processed_archive"),
)
def compute(ctx, source, output):
    if ctx.is_incremental and source.polars().is_empty():
        ctx.abort_job()
        return
    output.write_table(source.polars(lazy=True))

또는 실행 시 증분 구간에 파일이 하나 이상 포함되도록 구성합니다. 작업 전체 중단과 다음 실행의 입력 재처리는 트랜잭션 중단에서 설명합니다.

GPU 실행의 범위

cudf-polars를 설치하고 지원되는 NVIDIA GPU를 요청하면 lazy 쿼리를 collect(engine=pl.GPUEngine())로 GPU에서 실행할 수 있습니다. GPU 수집은 결과 전체를 메모리에 올리며 sink_parquet() 같은 스트리밍 쓰기는 지원하지 않습니다. 기본적으로 지원하지 않는 연산은 CPU로 되돌아가며, 이를 오류로 처리하려면 raise_on_fail=True를 사용합니다. 필요한 패키지와 메모리 조건은 GPU 설정과 예제를 따릅니다.

메모리를 넘는 입력, 선택적인 필터, 긴 연산 체인은 lazy 모드를 우선 검토합니다. 작은 데이터를 한 단계씩 탐색하거나 실행 시점을 분명히 보아야 하는 조사에는 eager 모드를 선택할 수 있습니다.