CSV와 JSON 파일에 스키마 적용하기

CSV나 JSON 파일을 데이터셋의 행과 열로 다루려면 스키마가 필요합니다. UI에서 파일 일부를 보고 스키마를 추론하여 적용하거나, 변환이 실행될 때 Spark가 파일을 읽으며 추론하도록 구성할 수 있습니다.

데이터셋 화면에서 적용

데이터셋의 Apply a schema를 선택하면 데이터 일부를 기준으로 스키마를 추론합니다. 이후 Edit schema에서 열 자료형과 추가 파싱 옵션을 바꿀 수 있습니다.

열 수가 맞지 않는 행을 제거하는 옵션
파일 인코딩
파일 경로, 행의 바이트 오프셋, 가져온 시각과 행 번호를 추가 열로 포함하는 옵션

초기 파일에 맞춰 정적으로 적용한 스키마는 이후 데이터 형식이 바뀌면 맞지 않을 수 있습니다. 입력 형식이 바뀌는 파이프라인에서는 첫 변환 단계에서 동적으로 추론하는 방법을 검토합니다.

동적 추론의 비용과 범위

Spark가 스키마를 추론하려면 입력을 읽어 자료형을 판단해야 하므로 빌드마다 추가 비용이 듭니다. 스키마가 바뀔 가능성이 있을 때 제한적으로 사용하고, 계약이 고정되어 있으면 StructType 또는 DDL 문자열로 스키마를 명시하는 편이 일관성과 성능에 유리합니다.

CSV에서 별도 스키마를 지정하지 않고 inferSchema=False로 읽으면 열을 문자열로 취급합니다. JSON에는 이 CSV 옵션을 그대로 적용하지 않습니다. JSON은 schema가 없으면 입력을 읽어 구조를 추론하므로, 추론을 피하려면 명시적 스키마를 전달합니다.

증분 배치마다 추론 결과가 달라지면 앞서 처리한 데이터와 새 데이터의 자료형이 충돌할 수 있습니다. 증분 파이프라인에서는 입력 스키마를 안정적으로 유지하는 것이 특히 중요합니다.

CSV 변환 예제

다음은 Python Spark Transforms 저장소에서 사용할 수 있는 파일 입력 변환입니다. 경로는 예시이며 실제 원시 CSV 데이터셋과 출력 경로로 바꿉니다. 입력에는 UTF-8 인코딩의 헤더가 있는 CSV 파일이 들어 있다고 가정합니다. 파일 목록 전체를 읽으므로 처리할 파일이 존재해야 합니다.

from transforms.api import Input, Output, transform
from transforms.verbs.dataframes import sanitize_schema_for_parquet


@transform(
    output=Output("/Company/sourceA/parsed/data_csv"),
    raw=Input("/Company/sourceA/raw/data_csv"),
)
def read_csv(ctx, raw, output):
    filesystem = raw.filesystem()
    hadoop_path = filesystem.hadoop_path
    files = [f"{hadoop_path}/{file.path}" for file in filesystem.ls()]
    dataframe = (
        ctx.spark_session.read
        .option("encoding", "UTF-8")
        .option("header", True)
        .option("inferSchema", True)
        .csv(files)
    )
    output.write_dataframe(sanitize_schema_for_parquet(dataframe))

UTF-8은 기본 인코딩이며 예제에서는 명시적으로 지정했습니다. header=True는 첫 행을 열 이름으로 사용합니다. 동적 추론이 필요하지 않으면 inferSchema를 끄거나, 해당 옵션 대신 입력 계약에 맞는 스키마를 reader의 schema 메서드에 전달합니다.

JSON 변환 예제

기본 JSON 입력은 각 줄이 독립적인 JSON 객체인 줄바꿈 구분 형식입니다. 파일 전체가 하나의 여러 줄 JSON 문서라면 multiLine=True로 읽어야 합니다. 다음 예제는 기본 줄바꿈 구분 형식을 사용합니다.

from transforms.api import Input, Output, transform
from transforms.verbs.dataframes import sanitize_schema_for_parquet


@transform(
    output=Output("/Company/sourceA/parsed/data_json"),
    raw=Input("/Company/sourceA/raw/data_json"),
)
def read_json(ctx, raw, output):
    filesystem = raw.filesystem()
    hadoop_path = filesystem.hadoop_path
    files = [f"{hadoop_path}/{file.path}" for file in filesystem.ls()]
    dataframe = (
        ctx.spark_session.read
        .option("multiLine", False)
        .json(files)
    )
    output.write_dataframe(sanitize_schema_for_parquet(dataframe))

multiLine=False는 기본값입니다. JSON 구조가 일정하다면 json 호출 전에 명시적인 스키마를 reader에 설정하여 매번 구조를 추론하지 않도록 할 수 있습니다. 두 예제는 서로 다른 출력 데이터셋을 사용합니다.

Parquet 출력의 열 이름

Transforms의 기본 출력 형식은 Parquet입니다. 자동 추론한 열 이름에 출력에서 허용되지 않는 특수 문자가 포함될 수 있으므로 예제에서는 sanitize_schema_for_parquet를 적용한 뒤 기록합니다. 출력의 열 이름이 소비자가 기대하는 스키마와 맞는지도 관리해야 합니다.

파일 전체나 일부를 SparkSession.read로 읽는 방식은 스키마 추론 외에도 사용할 수 있습니다. 동적 추론이 목적이 아니라면 고정 스키마로 같은 파일 접근 방식을 사용합니다.