Spark 파이프라인의 데이터량, 파티션과 조인 최적화

Spark 최적화는 필요한 결과를 유지하면서 읽기, 메모리와 데이터 이동을 줄이는 작업입니다. 자원 수를 먼저 늘리기보다 입력량, 연산 파티션과 출력 파일 배치를 구분하고 실행 지표에서 병목을 찾습니다.

먼저 줄일 데이터

필요하지 않은 행과 열을 가능한 한 이른 논리 단계에서 제외합니다. 필터는 구체적인 열과 상수 조건으로 표현하고, 같은 기능을 제공하는 DataFrame 내장 연산이 있으면 UDF보다 우선 검토합니다. 옵티마이저가 이해하기 쉬운 식은 조건 및 열 선택을 읽기 단계에 반영하는 데 유리합니다.

Spark는 선언한 연산을 최적화한 논리 및 물리 계획으로 바꾸므로 코드에 쓴 순서가 그대로 실행 순서라는 뜻은 아닙니다. 통계와 옵티마이저 설정에 따라 필터, 열 선택과 조인의 배치가 달라질 수 있습니다. 불필요한 정렬을 일찍 추가하기보다 결과에 필요한 연산인지 확인합니다.

가상 D 데이터셋을 SMALL과 BIG에 내부 조인할 때, SMALL과의 조인이 많은 행을 먼저 제거한다면 다음 논리 구조가 중간 데이터량을 줄일 수 있습니다.

SELECT D.key, D.key2
FROM D
JOIN SMALL ON SMALL.key = D.key
JOIN BIG ON BIG.key2 = D.key2

반대로 BIG을 먼저 결합하는 구조는 더 많은 중간 행을 처리할 수 있습니다.

SELECT D.key, D.key2
FROM D
JOIN BIG ON BIG.key2 = D.key2
JOIN SMALL ON SMALL.key = D.key

작은 테이블을 먼저 썼다는 이유만으로 항상 더 빠른 것은 아닙니다. 선택도, 중복 키, 통계와 실제 조인 계획을 확인합니다. 각 입력에서 사용하지 않는 열까지 제외하면 shuffle과 메모리 부담을 더 줄일 수 있습니다.

값 정리

문자열의 대소문자와 공백 차이를 업무 의미에 맞게 정규화합니다. 대소문자를 구분하는 비교에서는 Spark와 spark, filter와 Filter가 다른 값입니다. 둘을 같은 범주로 취급해야 하면 일관된 대소문자로 정리하거나 두 값을 명시적으로 포함하는 필터를 사용합니다.

저장 형식의 메타데이터와 파일 배치만으로 서로 다른 문자열의 업무상 의미를 자동으로 알아내지는 않습니다. 정리 규칙이 식별자나 다른 의미 있는 차이를 없애지 않는지도 확인합니다.

파티션 수를 정하는 기준

태스크 하나가 지나치게 많은 데이터를 처리하지 않도록 하면서, 너무 많은 작은 태스크와 파일의 시작 부담도 줄여야 합니다. 읽기 분할 수를 가늠하는 초기 기준은 다음과 같습니다.

이 추정치를 실행기 수와 반드시 같게 맞추는 규칙으로 사용하지 않습니다. 실행기 하나가 여러 태스크를 순차 또는 병렬로 처리할 수 있으며, 현대적인 파일 스캔은 형식과 읽기 설정도 고려합니다. 분할 불가능한 큰 파일은 한 읽기 단위에 부담을 집중시킬 수 있습니다.

입력 크기만 보고 나누어도 조인이나 집계 뒤에 특정 그룹이 커져 메모리가 부족할 수 있습니다. 다대다 조인은 입력의 합보다 큰 출력을 만들 수 있으므로 연산 이후의 분포까지 확인합니다.

기법별 선택

열의 고유값 수는 메모리, 키 쏠림과 출력 파일 수에 영향을 주지만 1,000개 같은 숫자를 모든 repartition 호출의 지원 한도로 해석하지 않습니다. 키를 지정한 연산 파티션과 키별 디렉터리를 만드는 Hive 파티션은 서로 다른 기능입니다.

Coalesce와 repartition

Coalesce는 기존 파티션을 합쳐 개수를 줄입니다. 보통 shuffle 없이 동작하므로 이동 비용이 작지만 기존 분포가 고르지 않으면 결과도 고르지 않습니다. 파티션을 하나로 크게 줄이면 계산이 소수 노드에 집중될 수 있습니다.

repartition(N)은 키를 지정하지 않고 데이터를 다시 분배합니다. repartition(N, key)는 키를 해시하여 같은 키의 행을 모으므로 키 분포가 쏠리면 결과도 쏠릴 수 있습니다. shuffle에는 네트워크 및 임시 디스크 비용이 들기 때문에 입력이 크면 이득과 비용을 비교합니다.

다음 Python 함수는 실제 Spark DataFrame과 양의 목표 파티션 수를 받아 해당 연산을 구성합니다.

from pyspark.sql import DataFrame


def reduce_partitions(dataframe: DataFrame, count: int) -> DataFrame:
    return dataframe.coalesce(count)


def redistribute(dataframe: DataFrame, count: int) -> DataFrame:
    return dataframe.repartition(count)


def group_by_key(dataframe: DataFrame, count: int, key: str) -> DataFrame:
    return dataframe.repartition(count, key)

Java에서도 같은 세 선택을 사용할 수 있습니다.

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

public final class Partitioning {
    private Partitioning() {}

    public static Dataset<Row> reduce(Dataset<Row> dataframe, int count) {
        return dataframe.coalesce(count);
    }

    public static Dataset<Row> redistribute(Dataset<Row> dataframe, int count) {
        return dataframe.repartition(count);
    }

    public static Dataset<Row> groupByKey(
            Dataset<Row> dataframe, int count, String key) {
        return dataframe.repartition(count, col(key));
    }
}

Spark SQL에서는 파티셔닝 힌트 또는 DISTRIBUTE BY를 사용할 수 있습니다. 다음은 실제 source_rows 테이블을 읽는 별도 선택지입니다.

SELECT /*+ COALESCE(20) */ * FROM source_rows;
SELECT /*+ REPARTITION(20) */ * FROM source_rows;
SELECT /*+ REPARTITION(20, column1) */ * FROM source_rows;
SELECT * FROM source_rows DISTRIBUTE BY column1;

DataFrame을 재파티션한 사실만으로 하위 작업이 파일을 읽을 때 그 배치를 알아내는 메타데이터가 저장되는 것은 아닙니다. 이를 활용하려면 출력의 버킷이나 Hive 파티션 구성을 함께 지정합니다.

버킷과 정렬

버킷은 키의 해시 결과를 고정된 수의 파일 그룹에 나눕니다. 하나의 버킷에는 여러 키 값이 들어갈 수 있습니다. 같은 키를 반복해서 조인하거나 집계하는 데이터에 적합하며 여러 입력을 같은 키와 호환되는 버킷 수로 작성하면 효과를 얻기 쉽습니다.

목표 파일 크기는 일반적으로 128~512MB를 검토할 수 있습니다. 그러나 입력 128MB가 연산 뒤 512MB가 될 수도 있으므로 입력 파일 크기만 기준으로 정하지 않습니다. 노드 메모리, 출력 확장과 태스크 수를 함께 비교합니다.

SQL의 저장 메타데이터와 shuffle

Code Repositories의 SQL 변환에서는 다음과 같이 버킷과 정렬 메타데이터를 지정할 수 있습니다. a, b, c를 가진 실제 입력 경로로 바꿉니다.

CREATE TABLE `/examples/optimization/bucketed_abc`
USING parquet
CLUSTERED BY (a)
SORTED BY (a)
INTO 200 BUCKETS AS
SELECT a, b, c
FROM `/examples/optimization/source_abc`
CLUSTER BY a

• CLUSTERED BY와 SORTED BY는 테이블의 파일 배치 및 정렬 메타데이터를 지정합니다.
• CLUSTER BY는 쓰기 전 데이터를 키별로 재분배하고 정렬하는 질의 연산입니다.

각 최종 태스크가 독립적으로 버킷 파일을 쓰므로 재분배 없이 쓰면 같은 버킷에 여러 파일이 생길 수 있습니다. 예를 들어 최종 태스크 200개가 각각 200개 버킷을 모두 건드리면 최대 40,000개 파일이 생기는 형태가 됩니다. 고유 키가 200개라는 사실만으로 200개 버킷이 모두 채워지는 것은 아닙니다.

앞선 집계 등으로 이미 키별 배치가 맞으면 추가 CLUSTER BY를 생략하여 shuffle을 줄일 수 있지만 실제 출력 파일 수를 확인해야 합니다. SQL에서도 항상 버킷당 파일 하나를 보장하는 것은 아닙니다. CREATE TABLE을 사용하는 이 SQL은 Code Workbook이 아니라 SQL Code Repository에서 사용합니다.

청구 데이터 예제

가상 claims는 100GB, patients는 1GB이며 patient_id로 자주 조인한다고 가정합니다.

SELECT claims.patient_id, patients.age
FROM claims
LEFT JOIN patients ON claims.patient_id = patients.patient_id

일반적인 sort-merge 계획은 양쪽 FileScan 뒤의 Exchange hashpartitioning과 Sort를 거쳐 SortMergeJoin을 수행합니다. 양쪽을 호환되는 버킷 및 정렬 구성으로 미리 쓰면 같은 데이터를 다시 이동하고 정렬하는 부담을 줄일 수 있습니다.

Python Code Repositories의 런타임 출력과 데이터프레임을 받는 쓰기 함수입니다.

def write_patient_buckets(output, dataframe):
    output.write_dataframe(
        dataframe.repartition(200, "patient_id"),
        bucket_cols=["patient_id"],
        bucket_count=200,
        sort_by=["patient_id"],
    )

Java에서는 같은 배치를 DatasetFormatSettings로 지정합니다.

import com.palantir.foundry.spark.api.DatasetFormatSettings;
import com.palantir.transforms.lang.java.api.FoundryOutput;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import static org.apache.spark.sql.functions.col;

public final class BucketWriter {
    private BucketWriter() {}

    public static void write(FoundryOutput output, Dataset<Row> dataframe) {
        DatasetFormatSettings format = DatasetFormatSettings.builder()
            .numBuckets(200)
            .addBucketColumns("patient_id")
            .addSortColumns("patient_id")
            .build();
        output.getDataFrameWriter(dataframe.repartition(200, col("patient_id")))
            .setFormatSettings(format)
            .write();
    }
}

Code Workbook의 Python 변환에서는 반환값에 배치 정보를 지정할 수 있습니다.

from vector.api import DataFrameReturn


def bucket_patient_rows(my_input):
    return DataFrameReturn(
        my_input.repartition(200, "patient_id"),
        bucket_cols=["patient_id"],
        bucket_count=200,
        sort_by=["patient_id"],
    )

SQL 변환의 해당 예시는 다음과 같습니다. claims_raw에 patient_id가 존재한다고 가정합니다.

CREATE TABLE `/examples/optimization/claims`
USING parquet
CLUSTERED BY (patient_id)
SORTED BY (patient_id)
INTO 200 BUCKETS AS
SELECT *
FROM `/examples/optimization/claims_raw`
CLUSTER BY patient_id

patients 쪽도 같은 키와 버킷 수로 작성합니다. 실제 소비자가 버킷 및 정렬 메타데이터를 활용하는 경우 FileScan에서 바로 조인으로 이어지고 불필요한 Exchange 및 Sort가 줄어듭니다. 엔진 설정과 입력 조건에 따라 정렬이 남을 수 있으므로 물리 계획으로 확인합니다.

쓰기 중 버킷 분할

쓰기 태스크는 개념적으로 각 행의 버킷 번호를 계산하고 버킷 및 정렬 열에 따라 묶어 파일을 만듭니다. 키 순서가 A, C, A, A, B, C이고 예시에서 A와 B가 버킷 0, C가 버킷 1이라면 다음처럼 나눕니다.

한 태스크에 두 버킷이 있으면 그 태스크는 최소 두 파일을 쓸 수 있습니다. 출력 직전에 호환되는 repartition(200, "patient_id")로 같은 버킷의 행을 모으면 중복 파일을 줄일 수 있습니다. 파일 분할 한도 등 다른 쓰기 설정도 실제 파일 수에 영향을 줍니다. 과도한 파일은 하위 읽기뿐 아니라 Catalog 같은 메타데이터 서비스에도 부담을 줍니다.

Hive 방식 파티셔닝

Hive 파티셔닝은 출력의 키 값을 디렉터리 이름에 기록합니다. 예를 들어 연도와 월로 나누면 year=2017/month=09 같은 경로가 됩니다. 해당 열의 필터에 맞는 디렉터리만 읽어 파일 수와 입력량을 줄일 수 있습니다.

한 태스크 안에서도 고유 키마다 파일을 쓰므로 고유값 수가 많으면 작은 파일이 늘어납니다. 이 방식 자체가 조인에 필요한 재분배를 항상 없애는 것은 아닙니다. 주된 이점인 파티션 가지치기와 조인의 버킷 메타데이터를 구분합니다.

SQL 쓰기

CREATE TABLE `/examples/optimization/hive_abc`
USING parquet
PARTITIONED BY (a) AS
SELECT a, b, c
FROM `/examples/optimization/source_abc`
CLUSTER BY a

쓰기 전의 키별 배치가 맞지 않으면 각 태스크가 같은 디렉터리에 파일을 따로 쓰므로 CLUSTER BY의 역할을 함께 고려합니다. CLUSTERED BY와 PARTITIONED BY를 결합하여 디렉터리 안에 버킷을 만들 수도 있지만 파일 수가 더 늘 수 있습니다. 이 CREATE TABLE 문법은 Code Workbook에서는 사용할 수 없습니다.

날짜별 로그 예제

로그의 날짜 열을 파티션으로 사용하면 다음과 같은 파일 경로를 만들 수 있습니다.

SELECT * FROM logs WHERE date < to_date('2017-01-03')

이 예시 범위에서는 1월 1일과 2일 디렉터리만 읽으면 됩니다. 물리 계획의 PartitionFilters에 날짜 조건이 있고 PartitionCount: 2처럼 선택한 Hive 파티션 수가 표시되는지 확인할 수 있습니다.

Python의 출력 쓰기 함수는 다음과 같습니다.

def write_logs_by_date(output, dataframe):
    output.write_dataframe(
        dataframe.repartitionByRange("date"),
        partition_cols=["date"],
    )

Java의 동일한 형식 설정은 다음과 같습니다.

import com.palantir.foundry.spark.api.DatasetFormatSettings;
import com.palantir.transforms.lang.java.api.FoundryOutput;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import static org.apache.spark.sql.functions.col;

public final class DatePartitionWriter {
    private DatePartitionWriter() {}

    public static void write(FoundryOutput output, Dataset<Row> dataframe) {
        DatasetFormatSettings format = DatasetFormatSettings.builder()
            .addPartitionColumns("date")
            .build();
        output.getDataFrameWriter(dataframe.repartitionByRange(col("date")))
            .setFormatSettings(format)
            .write();
    }
}

SQL에서는 다음처럼 실제 로그 열을 선택합니다.

CREATE TABLE `/examples/optimization/logs`
USING parquet
PARTITIONED BY (date) AS
SELECT content, date
FROM `/examples/optimization/logs_raw`
CLUSTER BY date

직접 소비자가 Mesa Transforms 데이터셋이면 Hive 파티셔닝을 사용하지 않습니다. 이 경로에서는 파티션 열을 올바르게 읽지 못하고 null로 해석할 수 있습니다. 파일 배치와 큰 키 그룹의 추가 조건은 Hive 파티셔닝에서 다룹니다.

조인과 집계 결과를 재사용

같은 결합이나 집계를 여러 하위 변환 및 Contour 분석이 반복한다면 별도 데이터셋으로 구체화할 수 있습니다. 한 번 계산한 결과를 재사용하여 반복 shuffle을 줄이는 방식입니다.

다만 필터링하지 않은 조인 결과가 너무 크거나 사용 빈도가 낮으면 추가 저장 및 파이프라인 단계의 비용이 더 클 수 있습니다. 구체화한 결과를 읽는 하위 작업도 자신의 연산에 따라 새로운 shuffle이 필요할 수 있습니다.

Sort-merge와 broadcast 조인

자주 고려하는 두 전략은 sort-merge와 broadcast입니다. Spark에는 다른 조인 전략도 있으며 실제 선택은 키, 조인 유형, 통계와 설정에 따라 달라집니다.

• Sort-merge는 필요할 때 양쪽 데이터를 키별로 shuffle하고 각 파티션을 정렬한 뒤 병합합니다. 호환되는 버킷 및 정렬이 있으면 그 비용을 줄일 수 있습니다.
• Broadcast는 작은 쪽 데이터를 필요한 실행기에 복제하여 다른 쪽의 행별 shuffle을 줄입니다. 복제본을 수용할 메모리가 각 실행기에 필요합니다.

작은 쪽이 어느 입력인지는 조인 유형과 통계에 따라 정합니다. 왼쪽 외부 조인에서 작은 오른쪽 참조 데이터를 복제하는 예가 흔하지만 항상 오른쪽만 복제하는 일반 규칙은 아닙니다.

from pyspark.sql import DataFrame, functions as F


def join_small_reference(events: DataFrame, reference: DataFrame) -> DataFrame:
    return events.join(
        F.broadcast(reference), events["key"] == reference["key"], "left"
    )

SQL의 같은 힌트 예시입니다.

SELECT /*+ BROADCAST(b) */ a.*, b.label
FROM `/examples/optimization/events` AS a
LEFT JOIN `/examples/optimization/reference` AS b
ON a.key = b.key

reference에는 key와 label이 있고 전체 데이터가 실행기 메모리에 충분히 들어간다고 가정합니다. 백만 행 미만 같은 행 수는 규모를 짐작하는 참고일 뿐, 넓은 문자열 열과 메모리 표현을 무시한 안전 한도가 아닙니다.

spark.sql.autoBroadcastJoinThreshold의 기본값은 10MB(10,485,760바이트)입니다. 옵티마이저가 가진 테이블 크기 통계와 해당 설정에 따라 자동 브로드캐스트를 선택할 수 있으며 -1로 비활성화할 수 있습니다. 디스크 압축 파일 크기가 작다는 사실만으로 실행 메모리도 작다고 보장하지 않습니다. 명시적 힌트는 임계값보다 큰 추정값에도 전략을 제안할 수 있지만 모든 조인 유형에서 적용된다는 보장은 없습니다.

변환 분리와 자원 확대

기존 기법으로 병목을 해결하기 어렵거나 중간 상태를 따로 관리할 필요가 있으면 변환을 나눌 수 있습니다. 단계별 최소 입력과 최적 배치를 명확히 할 수 있지만 실행 준비, 스케줄링과 오케스트레이션 비용이 추가됩니다.

자원을 더 요청하기 전후에는 같은 업무 결과를 기준으로 시간과 사용량을 비교합니다. 큰 자원을 과도하게 요청하는 작업이 여러 개면 공유 자원 확보와 반환 과정이 다른 작업에도 영향을 줄 수 있습니다. 필요한 성능을 달성할 다른 방법이 없으면 관리자 또는 Palantir 지원과 적절한 자원 할당을 결정합니다.

관련 실행 설정

관리형 프로필은 과거 사용량으로 할당을 조정하고, 네이티브 가속은 지원 연산을 Velox로 실행합니다. 프로필 적용과 전체 설정 참조에서 정확한 구성을 확인하고, 사용량 최적화에서 실행 주기와 전체 비용의 우선순위를 함께 검토합니다.