Partitioning

Partitioning(파티셔닝)은 대용량 테이블이나 인덱스를 논리적 기준에 따라 여러 개의 작은 단위로 분할하여 저장하는 데이터베이스 기법입니다. 수억 건 이상의 데이터를 단일 테이블에서 관리하면 쿼리 성능 저하, 인덱스 비대화, 관리 복잡성 증가 등의 문제가 발생하는데, 파티셔닝을 통해 이러한 문제를 해결할 수 있습니다. 논리적으로는 하나의 테이블이지만 물리적으로는 여러 파티션으로 저장되어 투명하게 접근할 수 있습니다.

Range Partitioning(범위 파티셔닝)은 날짜, 숫자 등 연속된 값의 범위를 기준으로 데이터를 분할합니다. 시계열 데이터에 적합하며, 예를 들어 월별로 파티션을 생성하면 특정 기간 조회 시 해당 파티션만 스캔하여 성능이 크게 향상됩니다. Hash Partitioning(해시 파티셔닝)은 특정 컬럼의 해시값을 기준으로 데이터를 균등하게 분산시키며, 데이터 분포가 불규칙한 경우에 효과적입니다. List Partitioning(리스트 파티셔닝)은 특정 값 목록을 기준으로 분할하여 지역 코드나 상태값 같은 명시적 그룹화에 사용됩니다.

Composite Partitioning(복합 파티셔닝)은 두 가지 이상의 파티셔닝 방식을 조합합니다. 예를 들어 Range-Hash는 먼저 날짜로 범위 파티션을 만들고, 각 파티션 내에서 해시로 서브파티션을 생성합니다. 이를 통해 시간 기반 쿼리와 균등 분산의 장점을 모두 얻을 수 있습니다. 또한 Partition Pruning(파티션 프루닝)은 쿼리 조건에 해당하는 파티션만 스캔하는 최적화 기법으로, 파티셔닝의 핵심 이점입니다.

파티셔닝과 Sharding(샤딩)은 종종 혼동되지만 다른 개념입니다. 파티셔닝은 단일 데이터베이스 내에서 테이블을 분할하는 것이고, 샤딩은 여러 데이터베이스 서버에 데이터를 분산하는 것입니다. 현대 분산 데이터베이스(Vitess, CockroachDB, TiDB)는 두 개념을 결합하여 수평적 확장성을 제공합니다. AI/ML 워크로드에서 대용량 학습 데이터셋을 효율적으로 저장하고 조회하기 위해 파티셔닝은 필수적인 기법이며, 데이터 레이크에서도 Hive 스타일 파티셔닝이 널리 사용됩니다.

PostgreSQL Range Partitioning

-- 부모 테이블 생성 (파티션 정의)
CREATE TABLE orders (
    id BIGSERIAL,
    order_date DATE NOT NULL,
    customer_id INTEGER,
    amount DECIMAL(10, 2),
    status VARCHAR(20)
) PARTITION BY RANGE (order_date);

-- 월별 파티션 생성
CREATE TABLE orders_2024_01 PARTITION OF orders
    FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');

CREATE TABLE orders_2024_02 PARTITION OF orders
    FOR VALUES FROM ('2024-02-01') TO ('2024-03-01');

CREATE TABLE orders_2024_03 PARTITION OF orders
    FOR VALUES FROM ('2024-03-01') TO ('2024-04-01');

-- 기본 파티션 (범위에 포함되지 않는 데이터용)
CREATE TABLE orders_default PARTITION OF orders DEFAULT;

-- 파티션별 인덱스 자동 생성
CREATE INDEX idx_orders_customer ON orders (customer_id);
CREATE INDEX idx_orders_status ON orders (status);

-- 쿼리 시 파티션 프루닝 확인
EXPLAIN ANALYZE
SELECT * FROM orders
WHERE order_date BETWEEN '2024-02-01' AND '2024-02-29';
-- orders_2024_02 파티션만 스캔됨

-- 파티션 분리 (오래된 데이터 아카이브)
ALTER TABLE orders DETACH PARTITION orders_2024_01;
-- 독립 테이블로 변환되어 아카이브 가능

MySQL Hash Partitioning

-- 해시 파티셔닝으로 균등 분산
CREATE TABLE user_sessions (
    session_id BIGINT NOT NULL,
    user_id INT NOT NULL,
    created_at DATETIME,
    data JSON,
    PRIMARY KEY (session_id, user_id)
)
PARTITION BY HASH(user_id)
PARTITIONS 8;

-- 리스트 파티셔닝 (지역별)
CREATE TABLE customers (
    id INT NOT NULL,
    name VARCHAR(100),
    region VARCHAR(20),
    PRIMARY KEY (id, region)
)
PARTITION BY LIST COLUMNS(region) (
    PARTITION p_asia VALUES IN ('KR', 'JP', 'CN', 'TW'),
    PARTITION p_europe VALUES IN ('DE', 'FR', 'UK', 'IT'),
    PARTITION p_america VALUES IN ('US', 'CA', 'BR', 'MX'),
    PARTITION p_others VALUES IN (DEFAULT)
);

-- Range-Hash 복합 파티셔닝
CREATE TABLE access_logs (
    id BIGINT NOT NULL,
    access_time DATETIME NOT NULL,
    user_id INT,
    endpoint VARCHAR(255),
    PRIMARY KEY (id, access_time, user_id)
)
PARTITION BY RANGE (YEAR(access_time))
SUBPARTITION BY HASH(user_id)
SUBPARTITIONS 4 (
    PARTITION p2023 VALUES LESS THAN (2024),
    PARTITION p2024 VALUES LESS THAN (2025),
    PARTITION p2025 VALUES LESS THAN (2026)
);

-- 파티션 정보 조회
SELECT TABLE_NAME, PARTITION_NAME, TABLE_ROWS, DATA_LENGTH
FROM INFORMATION_SCHEMA.PARTITIONS
WHERE TABLE_NAME = 'access_logs';

Python + Apache Spark 파티셔닝

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, year, month

spark = SparkSession.builder \
    .appName("DataPartitioning") \
    .getOrCreate()

# 대용량 데이터 로드
df = spark.read.parquet("s3://datalake/raw/events/")

# Hive 스타일 파티셔닝으로 저장
df.write \
    .partitionBy("year", "month", "day") \
    .mode("overwrite") \
    .parquet("s3://datalake/partitioned/events/")

# 결과 디렉토리 구조:
# s3://datalake/partitioned/events/
# ├── year=2024/
# │   ├── month=01/
# │   │   ├── day=01/
# │   │   │   └── part-00000.parquet
# │   │   └── day=02/
# │   └── month=02/
# └── year=2025/

# 파티션 프루닝이 적용된 읽기
df_january = spark.read.parquet("s3://datalake/partitioned/events/") \
    .where(col("year") == 2024) \
    .where(col("month") == 1)
# year=2024/month=01 파티션만 스캔

# Dynamic Partition Overwrite
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
df_updates.write \
    .partitionBy("year", "month") \
    .mode("overwrite") \
    .parquet("s3://datalake/partitioned/events/")
# 업데이트된 파티션만 덮어쓰기

# Repartitioning (메모리 내 재분할)
df_balanced = df.repartition(100, col("customer_id"))
df_coalesced = df.coalesce(10)  # 파티션 수 줄이기

# Bucket을 사용한 조인 최적화
df.write \
    .bucketBy(32, "customer_id") \
    .sortBy("order_date") \
    .saveAsTable("orders_bucketed")

자동 파티션 관리 스크립트

import psycopg2
from datetime import datetime, timedelta

def manage_partitions(conn, table_name: str, months_ahead: int = 3, months_retain: int = 24):
    """
    파티션 자동 생성 및 정리

    - 미래 N개월 파티션 미리 생성
    - 보관 기간 초과 파티션 분리/삭제
    """
    cursor = conn.cursor()
    today = datetime.now()

    # 미래 파티션 생성
    for i in range(months_ahead):
        target_date = today + timedelta(days=30 * i)
        partition_name = f"{table_name}_{target_date.strftime('%Y_%m')}"
        start_date = target_date.replace(day=1)
        end_date = (start_date + timedelta(days=32)).replace(day=1)

        create_sql = f"""
        CREATE TABLE IF NOT EXISTS {partition_name}
        PARTITION OF {table_name}
        FOR VALUES FROM ('{start_date.strftime('%Y-%m-%d')}')
        TO ('{end_date.strftime('%Y-%m-%d')}');
        """

        try:
            cursor.execute(create_sql)
            print(f"Created partition: {partition_name}")
        except psycopg2.errors.DuplicateTable:
            pass  # 이미 존재

    # 오래된 파티션 분리
    cutoff_date = today - timedelta(days=30 * months_retain)

    cursor.execute(f"""
        SELECT schemaname, tablename
        FROM pg_tables
        WHERE tablename LIKE '{table_name}_%'
    """)

    for schema, partition in cursor.fetchall():
        try:
            partition_date = datetime.strptime(
                partition.replace(f"{table_name}_", ""),
                "%Y_%m"
            )
            if partition_date < cutoff_date:
                cursor.execute(f"ALTER TABLE {table_name} DETACH PARTITION {partition}")
                cursor.execute(f"DROP TABLE {partition}")
                print(f"Dropped old partition: {partition}")
        except ValueError:
            continue

    conn.commit()

# 매일 실행하여 파티션 관리
if __name__ == "__main__":
    conn = psycopg2.connect("postgresql://localhost/mydb")
    manage_partitions(conn, "orders", months_ahead=3, months_retain=24)