1. Horizontal Partitioner#

01. Pattern Overview#

• 수평 파티셔닝은 행(row) 단위로 데이터를 물리적으로 분리된 저장 공간에 나누는 패턴.
• 특정 속성(distribution key)의 값에 따라 전체 행이 해당 파티션으로 이동
• 가장 기본적인 목적은 쿼리시 불필요한 데이터를 읽지 않는것 (partition pruning)

02. Distribution Key 선택#

파티셔닝 키는 두 가지 출처를 가질 수있음

• Job execution context: 파이프라인 실행 시점 기준. 같은 실행의 모든 레코드가 동일 파티션에 속함

• Dataset (event time): 데이터 자체의 이벤트 시간 기준. late data로 인해 하나의 배치에서 여러 파티션에 데이터가 분산 가능

• 시간 외에도 customer ID, 지역 등 도메인 비즈니스 키로 파티셔닝 가능

• 파티셔닝 키는 반드시 low-cardinality 속성이어야 함

• 고카디널리티 값을 쓰면 수백만 개의 작은 파티션이 생겨 small files 문제 발생

03. 선언적 vs 동적 파티셔닝#

• 선언적(Declarative): CREATE TABLE ... PARTITIONED BY로 테이블 생성 시 정의. 데이터 producer가 파티셔닝 로직을 몰라도 가능 (BigQuery, Databricks)
• 동적(Dynamic): Apache Spark의 partitionBy처럼 producer가 직접 파티션 컬럼을 지정.

04. 파티셔닝의 결과들#

• 메타데이터 오버헤드: 파티션이 너무 많으면 listing 작업이 느려지고 small files 문제 발생

• data skew: 파티션 간 데이터 분포가 불균등하면 특정 파티션이 병목이 된다. 특히 microbatch 스트리밍에서 하나의 큰 파티션이 전체 배치를 블로킹한다. 이를 완화하려면 backpressure buffer로 초과 레코드를 다음 배치로 밀어내는 방식

Mutability: 파티션 키 변경은 전체 데이터 재배치가 필요해 매우 비싼연산임. Apache Iceberg는 메타데이터 레벨에서 파티션 evolution을 지원하지만, 기존 데이터의 물리적 위치는 변경하지 않는다.

05 파티셔닝 vs 샤딩#

• 샤딩은 데이터를 물리적으로 다른 머신에 분산하는 것
• 수평 파티셔닝은 논리적 분리이며 반드시 머신 간 이동을 수반하지 않음 -> 즉, 샤딩은 수평 파티셔닝의 보다 특수한 형태

1
from pyspark.sql import SparkSession, functions
2

3
spark = SparkSession.builder.getOrCreate()
4

5
input_users = spark.read.schema("user_id STRING, change_date TIMESTAMP, name STRING") \
6
    .json("/data/users")
7

8
# 시간 기반 중첩 파티셔닝 컬럼 생성
9
partitioned_users = (
10
    input_users
11
    .withColumn("year", functions.year("change_date"))
12
    .withColumn("month", functions.month("change_date"))
13
    .withColumn("day", functions.day("change_date"))
14
    .withColumn("hour", functions.hour("change_date"))
15
)
16

17
# year/month/day/hour 기반 파티셔닝으로 저장
18
(
19
    partitioned_users.write
20
    .mode("overwrite")
21
    .format("parquet")
22
    .partitionBy("year", "month", "day", "hour")
23
    .save("/output/users_partitioned")
24
)