Hive 방식으로 데이터셋 파일 나누기

Hive-style partitioning은 특정 열의 값을 파일 경로에 기록하여, 그 열로 필터링할 때 읽을 파일을 줄이는 저장 방식입니다. Spark 데이터프레임의 연산 파티션과 파일 배치를 위한 Hive 파티션은 구분해야 합니다.

쓰기와 읽기 동작

Spark 기반 변환이 출력을 쓸 때는 각 데이터프레임 파티션 안에서 지정한 파티션 열 값의 조합별로 파일을 만듭니다. 경로에는 해당 열 값이 포함되고, 트랜잭션 메타데이터에도 어떤 열로 나누었는지 기록합니다.

Spark나 Polars 같은 읽기 엔진은 이 메타데이터와 경로를 이용하여 필터 조건에 필요한 파일만 선택할 수 있습니다. 반면 서로 다른 값의 조합이 지나치게 많으면 조합마다 최소 하나의 파일이 필요하여 쓰기와 읽기가 느려질 수 있습니다. 값별 행 수가 적고 고유값이 많은 열은 이 방식에 적합하지 않습니다.

쓰기 전에 연산 파티션 정리

같은 Hive 키가 여러 Spark 파티션에 흩어져 있으면 각 파티션에서 파일을 따로 만들 수 있습니다. 쓰기 전에 같은 열로 repartitionByRange를 수행하면 같은 조합을 한 연산 파티션에 모아 파일 수를 줄일 수 있습니다. 기본적인 파일 분할 조건에서는 조합당 파일 하나를 목표로 하지만 maxRecordsPerFile 같은 별도 파일 한도가 있으면 더 나뉠 수 있습니다.

repartitionByRange는 표본으로 값의 구간을 추정하여 데이터를 분산합니다. 낮은 고유값 수를 가진 열에서는 해시를 파티션 수로 나눈 나머지를 사용하는 repartition보다 균형을 잡기 쉬워 권장됩니다. 표본과 값의 분포에 따라 구간은 달라질 수 있으며 데이터 쏠림이 완전히 사라진다는 보장은 아닙니다.

Python 예제

입력에 record_date와 department가 있고 이 두 열로 자주 필터링한다고 가정합니다. 경로는 실제 입력 및 출력으로 지정합니다.

from transforms.api import Input, Output, transform


@transform(
    transform_output=Output("/examples/layout/hive"),
    transform_input=Input("/examples/layout/source"),
)
def write_hive_partitioned(transform_output, transform_input):
    dataframe = transform_input.dataframe().repartitionByRange(
        "record_date", "department"
    )
    transform_output.write_dataframe(
        dataframe,
        partition_cols=["record_date", "department"],
    )

Java 예제

자동 등록 대상인 myproject.datasets 패키지에 다음 클래스를 둡니다. 기존 파이프라인 정의가 이 패키지를 autoBindFromPackage로 등록하도록 구성되어 있어야 합니다.

package myproject.datasets;

import com.palantir.foundry.spark.api.DatasetFormatSettings;
import com.palantir.transforms.lang.java.api.Compute;
import com.palantir.transforms.lang.java.api.FoundryInput;
import com.palantir.transforms.lang.java.api.FoundryOutput;
import com.palantir.transforms.lang.java.api.Input;
import com.palantir.transforms.lang.java.api.Output;

import static org.apache.spark.sql.functions.col;

public final class HivePartitioningInJava {
    @Compute
    public void compute(
            @Input("/examples/layout/source") FoundryInput input,
            @Output("/examples/layout/java_hive") FoundryOutput output) {
        output.getDataFrameWriter(
                input.asDataFrame().read().repartitionByRange(
                    col("record_date"), col("department")))
            .setFormatSettings(DatasetFormatSettings.builder()
                .addPartitionColumns("record_date", "department")
                .build())
            .write();
    }
}

큰 키 그룹을 여러 연산 파티션에 나누기

기본 예제는 연산 파티션 수를 지정하지 않으므로 spark.sql.shuffle.partitions의 기본 구성을 사용합니다. 다음 상황을 주의합니다.

• 같은 파티션 열 조합은 한 Spark 파티션으로 모이므로 그 그룹이 크면 한 실행기에 부하가 집중될 수 있습니다.
• 고유 조합 수가 연산 파티션 수보다 많으면 한 파티션이 여러 조합을 처리해야 합니다. 개별 조합이 작더라도 합쳐진 처리량이 커질 수 있습니다.

record_timestamp처럼 같은 날짜 및 부서 안에서 값을 더 나눌 수 있는 열을 연산 파티션 키에 추가할 수 있습니다. 파일 경로는 여전히 department, record_date로만 나눕니다.

from transforms.api import Input, Output, transform


@transform(
    transform_output=Output("/examples/layout/hive_spread"),
    transform_input=Input("/examples/layout/source"),
)
def spread_large_groups(transform_output, transform_input):
    dataframe = transform_input.dataframe()
    group_count = (
        dataframe.select("department", "record_date").distinct().count()
    )
    partition_count = max(1, group_count * 8)
    spread = dataframe.repartitionByRange(
        partition_count, "department", "record_date", "record_timestamp"
    )
    transform_output.write_dataframe(
        spread,
        partition_cols=["department", "record_date"],
    )

조합 수의 8배를 연산 파티션 수로 정하여 조합당 대략 여러 파일로 나누는 예시입니다. 빈 입력에서도 파티션 수는 최소 하나로 유지합니다. 실제 파일 수는 표본과 분포에 따라 달라지며 모든 조합에 정확히 8개가 생기는 것은 아닙니다. 세 열의 값이 모두 같은 큰 그룹은 여전히 한쪽에 모일 수 있습니다.

distinct().count() 자체도 계산을 수행하므로 파일 수 감소와 추가 연산의 비용을 함께 비교합니다. 자동 컴팩션이나 여러 조회 패턴에 맞는 별도 배치가 필요하면 데이터셋 프로젝션을 검토합니다.