스키마 없는 CSV와 Excel 파일 처리

스키마 없는 lightweight 입력은 filesystem().ls()로 파일을 나열하고 open()으로 읽습니다. 입력을 .polars()나 .path()로 직접 읽는 표 경로와 구분합니다. 파일마다 필요한 파서로 처리한 뒤 실제 표 객체를 write_table()에 전달합니다.

CSV 파일의 증분 처리

다음 예제는 헤더가 있는 UTF-8 CSV를 읽어 모든 열을 문자열로 유지합니다. 파일별 자료형 추론 차이를 피하며, 합칠 파일들은 같은 열 구조를 가져야 합니다. 가상의 경로를 실제 입력과 출력으로 바꿉니다.

from concurrent.futures import ThreadPoolExecutor

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


@incremental()
@transform.using(
    source=Input("/Example Project/Inputs/raw_csv_files"),
    output=Output("/Example Project/Outputs/parsed_csv_files"),
)
def compute(ctx, source, output):
    filesystem = source.filesystem()
    files = list(filesystem.ls(glob="*.csv"))
    if not files:
        if ctx.is_incremental:
            ctx.abort_job()
            return
        raise ValueError("초기 출력의 열 구조를 결정할 CSV 파일이 필요합니다.")

    def read_file(status):
        with filesystem.open(status.path, "rb") as stream:
            return pl.read_csv(stream, has_header=True, infer_schema=False)

    with ThreadPoolExecutor(max_workers=4) as executor:
        frames = list(executor.map(read_file, files))
    output.write_table(pl.concat(frames, how="vertical"))

첫 실행에서는 입력 전체를 읽어 출력을 만들고, 이후 증분 실행에서는 추가된 파일을 처리합니다. 새 파일이 없으면 작업 전체를 중단하여 이전 출력을 유지합니다. 초기 입력이 비어 있으면 파일에서 schema를 알아낼 수 없으므로 명시적으로 실패합니다.

예제의 4는 파일 동시 처리 스레드 수이며 Foundry에 요청하는 CPU 코어 수와 별개입니다. 결과를 모두 메모리에 모아 합치므로 전체 파싱 결과 크기와 스레드 수를 함께 조정합니다.

Excel의 임의 접근 준비

Excel reader는 seek 가능한 파일이 필요합니다. Foundry 입력 스트림을 로컬 임시 파일로 복사하고 flush와 seek를 수행한 뒤 읽습니다. 다음 예제는 .xlsx 파일의 첫 워크시트를 읽으며 환경에 pandas, polars, pyarrow, openpyxl이 필요합니다. 다른 Excel 형식은 그 형식을 지원하는 reader 엔진과 의존성을 준비합니다.

같은 열 구조를 가진 파일들을 문자열 표로 합칩니다. 문자열 변환 후에도 결측값은 결측값으로 유지합니다.

from concurrent.futures import ThreadPoolExecutor
import shutil
import tempfile

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


def read_excel_file(filesystem, path):
    with filesystem.open(path, "rb") as source:
        with tempfile.TemporaryFile() as temporary:
            shutil.copyfileobj(source, temporary)
            temporary.flush()
            temporary.seek(0)
            frame = pd.read_excel(
                temporary, engine="openpyxl", dtype="string"
            )
    return pl.from_pandas(frame)


@transform.using(
    source=Input("/Example Project/Inputs/raw_excel_files"),
    output=Output("/Example Project/Outputs/excel_table"),
)
def compute(source, output):
    filesystem = source.filesystem()
    paths = [item.path for item in filesystem.ls(glob="*.xlsx")]
    if not paths:
        raise ValueError("출력의 열 구조를 결정할 XLSX 파일이 필요합니다.")

    def read_path(path):
        return read_excel_file(filesystem, path)

    with ThreadPoolExecutor(max_workers=4) as executor:
        frames = list(executor.map(read_path, paths))
    output.write_table(pl.concat(frames, how="vertical"))

동시 처리가 필요 없으면 compute 안의 executor 블록을 다음 한 문장으로 바꿉니다. 나머지 파서와 출력 코드는 같습니다.

frames = [read_excel_file(filesystem, path) for path in paths]

파일을 먼저 복사하므로 원본 스트림의 임의 접근 제약을 피할 수 있습니다. 대신 동시 파싱 결과를 담을 메모리와 각 임시 파일을 담을 로컬 디스크가 필요합니다. 표 스키마가 없는 입력을 처리하더라도 출력은 파싱한 실제 DataFrame에서 스키마를 얻습니다.