다운로드할 CSV 파일 준비하기

일반적인 데이터 내보내기는 Foundry의 Actions 메뉴나 해당 애플리케이션의 내보내기를 우선 사용합니다. 그 경로를 사용할 수 없을 때 Code Repositories나 Pipeline Builder에서 작은 CSV 출력 데이터셋을 준비할 수 있습니다.

범위와 자료형

먼저 조직의 데이터 내보내기 정책에 따라 해당 데이터와 사용 목적이 허용되는지 확인합니다. 필요한 행과 열만 남겨 크기를 줄입니다.

압축하지 않은 CSV 크기를 기본 HDFS 블록 크기인 128 MB보다 작게 만드는 것이 권장됩니다. 이는 CSV 형식의 최대 한도가 아니라 단일 파일 생성의 성능과 자원 사용을 위한 기준입니다. 큰 파일은 더 오래 걸리고 실행기 메모리나 디스크 사용이 늘 수 있습니다.

CSV에는 강제된 열 스키마가 없으므로 최종 열을 String으로 바꾸는 것이 권장됩니다. 특히 timestamp의 문자열 표현을 명확히 정해야 합니다.

PySpark 준비와 쓰기

다음은 가상 택시 데이터에서 vendor_id=CMT인 행을 고르고 medallion, tip_amount만 내보내는 예입니다. 두 자산 경로는 실제 프로젝트의 입출력으로 바꿉니다. 입력에는 이 세 열이 있어야 합니다.

from pyspark.sql import functions as F
from transforms.api import Input, Output, transform


@transform(
    output=Output("/Example Project/Exports/taxi_csv"),
    trips=Input("/Example Project/Inputs/taxi_trips"),
)
def compute(output, trips):
    selected = (
        trips.dataframe()
        .filter(F.col("vendor_id") == F.lit("CMT"))
        .select(
            F.col("medallion").cast("string").alias("medallion"),
            F.col("tip_amount").cast("string").alias("tip_amount"),
        )
    )

    row_count = selected.count()
    if row_count > 1000:
        selected = selected.sample(
            withReplacement=False,
            fraction=1000.0 / row_count,
            seed=0,
        )

    output.write_dataframe(
        selected.repartition(1),
        output_format="csv",
        options={"header": "true", "maxRecordsPerFile": "0"},
    )

1,000행보다 적거나 비어 있는 입력은 그대로 유지합니다. 큰 입력에서 사용하는 sample은 약 1,000행을 기대하는 확률 표본이며 정확히 1,000행이나 최대 1,000행을 보장하지 않습니다. 엄격한 내보내기 행 수 또는 크기 제한이 있으면 별도 제한 조건을 적용해야 합니다.

repartition(1)은 최종 쓰기 파티션을 하나로 모읍니다. coalesce(1)도 파티션을 줄이지만 같은 변환의 상위 필터까지 적은 작업으로 실행하게 만들 수 있습니다. 필터를 병렬로 실행한 뒤 합치려면 shuffle이 있는 repartition(1)을 사용합니다.

Pipeline Builder

동일한 필터, 열 선택과 String 변환을 구성한 뒤 Repartition data를 1로 설정합니다.

Builder의 단일 파티션 설정

출력의 write format을 CSV로 지정합니다. 다른 요구가 있으면 JSON, ORC, Parquet 또는 Avro 등 지원 형식을 선택할 수 있습니다.

Builder의 CSV 출력 형식

Builder UI는 Spark의 모든 writer 옵션을 노출하지는 않습니다. 필요한 형식 옵션이 있는지 확인해야 합니다.

SQL

다음은 이미 작게 정리한 가상 입력을 CSV로 쓰는 Spark SQL 변환입니다. 실제 프로젝트의 경로로 바꿉니다.

CREATE TABLE `/Example Project/Exports/sql_csv`
USING CSV
OPTIONS (header 'true', maxRecordsPerFile '0')
AS SELECT /*+ REPARTITION(1) */ *
FROM `/Example Project/Exports/prepared_input`

단일 파티션이고 추가 디렉터리 분할이나 파일별 행 수 제한이 없다면 한 CSV 데이터 파일을 만드는 데 사용할 수 있습니다. 데이터셋의 전체 파일 목록에는 데이터 외의 메타데이터 파일 등이 있을 수 있으므로 내려받을 데이터 파일을 확인합니다.

자원과 파일 개수

최종 쓰기에는 하나의 task가 사용되므로 병렬 쓰기의 처리량을 잃습니다. Spark writer는 행을 순차로 기록할 수 있어 전체 CSV를 반드시 한 번에 메모리에 담아야 하는 것은 아니지만, 상위 연산과 shuffle 및 writer가 필요한 자원은 충분해야 합니다.

단일 task가 처리하기 어려우면 Spark 프로필의 실행기 자원을 조정하거나 repartition에 1보다 큰 수를 사용합니다. 여러 파티션이면 여러 CSV 파일을 만들어 각각 내려받게 됩니다. maxRecordsPerFile처럼 한 파티션을 여러 파일로 나누는 옵션도 최종 파일 수에 영향을 줍니다.

파일 다운로드

데이터셋 빌드 후 Dataset 페이지의 Details에서 생성한 CSV를 선택하여 다운로드합니다. 실제 다운로드에는 해당 자산의 권한과 내보내기 통제가 적용됩니다.

생성한 CSV 파일 다운로드

Keyboard shortcuts