데이터셋의 원시 파일 읽기와 쓰기

데이터셋 파일시스템은 저장소의 실제 위치를 몰라도 데이터셋 안의 논리 경로로 파일을 열게 합니다. XML, JSON, CSV나 압축 파일을 직접 처리할 때 사용합니다. 입력 핸들은 읽기용 파일시스템을, 출력 핸들은 쓰기용 파일시스템을 제공합니다.

핸들을 받는 transform(), transform.using() 또는 transform.spark.using()으로 변환을 선언합니다. DataFrame만 받는 decorator와 구분합니다. Spark의 FileSystem과 경량 실행의 FoundryDataSidecarFileSystem은 서로 다른 인터페이스이므로 같은 메서드 이름만 보고 모든 기능을 공통으로 가정하지 않습니다.

데이터셋으로 가져온 파일

수동 가져오기나 Data Connection 동기화로 파일을 데이터셋에 넣을 수 있습니다. CSV를 가져올 때는 Apply a schema로 표 스키마를 적용할 수도 있습니다. 다음 예제는 파일 자산 자체를 가리키는 것이 아니라, 파일을 담은 데이터셋을 입력으로 사용합니다. 원래 확장자를 유지한 원시 파일 업로드와 데이터셋의 파일 목록을 구분합니다.

Lightweight에서는 스키마 없는 데이터셋도 filesystem().ls()와 open()으로 처리할 수 있습니다. .path(), .pandas(), .polars()와 .arrow()는 스키마를 가진 입력을 대상으로 합니다. 경량 파일시스템에는 Spark의 분산 files() 메서드가 정의되어 있지 않습니다.

파일 목록과 메타데이터

ls()는 FileStatus를 내보냅니다.

가상의 입력 경로를 실제 파일 데이터셋으로 바꾸면 다음 변환이 CSV 파일 목록을 표로 만듭니다. 파일이 없을 때도 메타데이터 표의 schema를 명시합니다.

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


@transform.using(
    source=Input("/Example Project/Inputs/raw_files"),
    output=Output("/Example Project/Outputs/csv_file_inventory"),
)
def compute(source, output):
    statuses = source.filesystem().ls(glob="*.csv")
    rows = [
        {"path": item.path, "size": item.size, "modified": item.modified}
        for item in statuses
    ]
    output.write_table(pl.DataFrame(
        rows,
        schema={"path": pl.String, "size": pl.Int64, "modified": pl.Int64},
    ))

ls(glob="*.csv")는 glob 패턴으로, ls(regex=r"[A-Z]*\.csv")는 정규식으로 제한합니다. 두 번째 패턴의 대문자 파일 이름 조건에는 students.csv가 맞지 않습니다. 숨김 파일은 기본적으로 제외하며 필요한 경우 show_hidden=True를 지정합니다.

순차 읽기

open(path, mode="r", **kwargs)는 파일 유사 스트림을 반환하며 io.open()의 인자를 받습니다. seek()와 tell()을 전제로 하는 임의 접근은 사용할 수 없습니다. with 문으로 수명을 관리하고 바이너리는 rb, 텍스트는 문자 인코딩을 지정합니다.

다음 함수의 source는 변환에 전달된 실제 입력 핸들입니다. 입력에는 students.csv가 있고 첫 줄이 id,hair,eye,sex라고 가정합니다.

import csv


def read_first_line(source):
    with source.filesystem().open("students.csv", "r", encoding="utf-8") as stream:
        return stream.readline()


def read_csv_header(source):
    with source.filesystem().open(
        "students.csv", "r", encoding="utf-8", newline=""
    ) as stream:
        return next(csv.reader(stream, delimiter=","))

첫 함수는 줄바꿈을 포함한 첫 줄을 반환하고, 두 번째는 네 열 이름의 목록을 반환합니다. CSV의 인용 부호, 쉼표와 줄바꿈은 CSV 파서로 처리합니다.

임의 접근이 필요할 때

파일을 메모리나 로컬 디스크에 먼저 복사하면 임의 접근이 가능합니다. 전체 복사 비용이 생기므로 순차 파서로 처리할 수 있으면 그 방식을 우선 사용합니다.

작은 텍스트를 메모리에 복사하는 방법입니다.

import io
import shutil


def read_all_text(source):
    with io.StringIO() as buffer:
        with source.filesystem().open("students.csv", "r", encoding="utf-8") as stream:
            shutil.copyfileobj(stream, buffer)
        return buffer.getvalue()


def read_lines(source):
    with source.filesystem().open("students.csv", "r", encoding="utf-8") as stream:
        return stream.read().splitlines()

read_lines()는 줄바꿈을 제거한 문자열 목록을 만듭니다. 두 방식 모두 파일 전체가 메모리에 들어가야 합니다.

디스크에 복사한 뒤 다시 여는 방식도 사용할 수 있습니다.

import shutil
import tempfile


def read_local_first_line(source):
    with tempfile.NamedTemporaryFile() as temporary:
        with source.filesystem().open("students.csv", "rb") as stream:
            shutil.copyfileobj(stream, temporary)
        temporary.flush()
        with open(temporary.name, "r", encoding="utf-8") as local:
            return local.readline()

복사 후 flush를 호출하여 쓰기 버퍼를 반영하고, 임시 파일은 사용이 끝나면 정리합니다. ZIP과 Excel처럼 seek가 필요한 파서는 같은 원칙으로 임시 파일을 사용합니다.

파일 처리의 실행 방식

• ZIP 안의 CSV 읽기는 네 엔진에서 seek 가능한 압축 파일과 CSV 파서를 사용하는 예제입니다.
• CSV와 모델 파일 쓰기는 원시 파일 출력과 표 출력의 차이를 설명합니다.
• 스레드 및 Spark 분산 처리는 단일 노드의 스레드와 executor 계산을 구분합니다.
• 스키마 없는 CSV와 Excel 처리는 증분 빈 입력과 임시 파일이 필요한 스프레드시트를 다룹니다.