Hive Non-ACID ORC 테이블 Compaction 가이드

dt=2026-10-01
  2 MB, 3 MB, 1 MB, 4 MB, 2 MB, ...   → 1,000 files

SHOW TBLPROPERTIES dw.sales('transactional');

hdfs dfs -ls <partition 경로>

ALTER TABLE dw.sales PARTITION (dt='2026-10-01') CONCATENATE;

SET hive.merge.tezfiles=true;              -- Tez 엔진
SET hive.merge.mapfiles=true;
SET hive.merge.mapredfiles=true;
SET hive.merge.smallfiles.avgsize=134217728;   -- 평균 128 MB 미만이면 병합
SET hive.merge.size.per.task=536870912;        -- 병합 결과 목표 512 MB

CREATE TABLE dw.sales (
    id          BIGINT,
    customer_id STRING,
    amount      DECIMAL(18,2)
)
PARTITIONED BY (dt STRING)
STORED AS ORC;

/user/hive/warehouse/dw.db/sales/
├── dt=2026-09-29
├── dt=2026-09-30
└── dt=2026-10-01
      ├── 000000_0
      ├── 000001_0
      ├── ...
      └── 000999_0

Cannot overwrite a path that is also being read from.

Target Files = ceil(Partition Size / Target File Size)
 
예) 10 GB / 512 MB = 10 × 1024 / 512 = 20

#!/usr/bin/env python3
"""Hive Non-ACID ORC Partition Small File Compaction (PySpark)"""
 
import argparse
import math
import uuid
from datetime import datetime
 
from pyspark.sql import SparkSession
 
spark = (
    SparkSession.builder
    .appName("HiveORCPartitionCompaction")
    .enableHiveSupport()
    .getOrCreate()
)
 
 
# ------------------------------------------------------------
# Utility
# ------------------------------------------------------------
 
def sql_quote(value):
    return str(value).replace("'", "''")
 
 
def parse_partition_args(partition_args):
    """--partition dt=2026-10-01 (복수 지정 가능) → dict"""
    result = {}
    for item in partition_args:
        if "=" not in item:
            raise ValueError(f"잘못된 partition 형식: {item}")
        key, value = item.split("=", 1)
        key, value = key.strip(), value.strip()
        if not key:
            raise ValueError(f"Partition column이 비어 있습니다: {item}")
        result[key] = value
    return result
 
 
def make_partition_sql(partition_spec):
    """{"dt": "2026-10-01"} → `dt`='2026-10-01'"""
    return ", ".join(f"`{k}`='{sql_quote(v)}'" for k, v in partition_spec.items())
 
 
def make_where_sql(partition_spec):
    return " AND ".join(f"`{k}`='{sql_quote(v)}'" for k, v in partition_spec.items())
 
 
def human_size(size):
    value = float(size)
    for unit in ["B", "KB", "MB", "GB", "TB"]:
        if value < 1024:
            return f"{value:.2f} {unit}"
        value /= 1024
    return f"{value:.2f} PB"
 
 
# ------------------------------------------------------------
# Table Safety Check
# ------------------------------------------------------------
 
def validate_table(table_name):
    """
    Spark 방식으로 처리하면 안 되는 테이블을 사전에 차단한다.
      1. Hive ACID (transactional=true)
      2. Hive Bucketed Table (CLUSTERED BY)
 
    Spark 3.x의 SHOW CREATE TABLE은 Hive DDL이 아닌 Spark DDL을
    출력하고 Transactional Table에서는 예외를 내므로,
    Metastore 정보를 직접 조회한다.
    """
    properties = {
        str(row[0]).strip(): str(row[1]).strip()
        for row in spark.sql(f"SHOW TBLPROPERTIES {table_name}").collect()
    }
    if properties.get("transactional", "").lower() == "true":
        raise RuntimeError(
            "Hive ACID Transactional Table입니다. "
            "Spark Merge 대신 Hive MAJOR COMPACTION을 사용하십시오."
        )
 
    for row in spark.sql(f"DESCRIBE FORMATTED {table_name}").collect():
        key = str(row[0]).strip() if row[0] is not None else ""
        if key == "Num Buckets":
            value = str(row[1]).strip()
            if value.lstrip("-").isdigit() and int(value) > 0:
                raise RuntimeError(
                    "Hive Bucketed Table입니다. "
                    "CLUSTERED BY 테이블에는 이 프로그램을 사용하지 마십시오."
                )
    print("Table validation OK")
 
 
def validate_partition_spec(table_name, partition_spec):
    """
    모든 Partition Column이 지정되었는지 확인한다.
    year/month/day 테이블에 year=2026만 지정하면 나머지 Partition
    Column이 data column처럼 처리되므로 실행을 중단한다.
    """
    partition_columns = [
        c.name for c in spark.catalog.listColumns(table_name) if c.isPartition
    ]
    if not partition_columns:
        raise RuntimeError(f"Partition Table이 아닙니다: {table_name}")
    if {c.lower() for c in partition_columns} != {c.lower() for c in partition_spec}:
        raise ValueError(
            "모든 Partition Column을 지정해야 합니다. "
            f"테이블 Partition Column: {partition_columns}, "
            f"입력값: {list(partition_spec.keys())}"
        )
    return partition_columns
 
 
# ------------------------------------------------------------
# Partition Location / HDFS
# ------------------------------------------------------------
 
def get_partition_location(table_name, partition_spec):
    rows = spark.sql(
        f"DESCRIBE FORMATTED {table_name} "
        f"PARTITION ({make_partition_sql(partition_spec)})"
    ).collect()
    for row in rows:
        key = str(row[0]).strip() if row[0] is not None else ""
        if key == "Location":
            return str(row[1]).strip()
    raise RuntimeError("Partition Location을 찾을 수 없습니다.")
 
 
def _hdfs(path_string):
    jvm = spark.sparkContext._jvm
    conf = spark.sparkContext._jsc.hadoopConfiguration()
    path = jvm.org.apache.hadoop.fs.Path(path_string)
    return path.getFileSystem(conf), path
 
 
def get_hdfs_statistics(path_string):
    fs, path = _hdfs(path_string)
    if not fs.exists(path):
        raise RuntimeError(f"HDFS Path가 존재하지 않습니다: {path_string}")
    summary = fs.getContentSummary(path)
    return {"bytes": summary.getLength(), "files": summary.getFileCount()}
 
 
def delete_hdfs_path(path_string):
    fs, path = _hdfs(path_string)
    if fs.exists(path):
        fs.delete(path, True)
 
 
def calculate_target_files(partition_size_bytes, target_file_size_mb):
    target_bytes = target_file_size_mb * 1024 * 1024
    return max(1, math.ceil(partition_size_bytes / target_bytes))
 
 
# ------------------------------------------------------------
# Main Compaction
# ------------------------------------------------------------
 
def compact_partition(
    table_name,
    partition_spec,
    staging_base,
    target_file_size_mb=512,
    min_file_count=100,
    target_files_override=None,
    merge_mode="coalesce",
    keep_staging=False,
):
    print(f"Table          : {table_name}")
    print(f"Partition      : {partition_spec}")
 
    # 1. Table / Partition 검증
    validate_table(table_name)
    partition_columns = validate_partition_spec(table_name, partition_spec)
 
    # 2. Partition Location 및 HDFS 통계
    partition_location = get_partition_location(table_name, partition_spec)
    stats = get_hdfs_statistics(partition_location)
    partition_size, current_files = stats["bytes"], stats["files"]
    print(f"Location       : {partition_location}")
    print(f"Partition Size : {human_size(partition_size)}")
    print(f"Current Files  : {current_files}")
 
    # 3. Compaction 필요 여부
    if current_files < min_file_count:
        print(f"파일 수가 {min_file_count}개 미만이므로 Compaction을 수행하지 않습니다.")
        return
 
    # 4. Target File 개수 계산
    if target_files_override is not None:
        if target_files_override < 1:
            raise ValueError("--target-files는 1 이상이어야 합니다.")
        target_files = target_files_override
    else:
        target_files = calculate_target_files(partition_size, target_file_size_mb)
    print(f"Target Size    : {target_file_size_mb} MB")
    print(f"Target Files   : {target_files}")
 
    if current_files <= target_files:
        print("현재 파일 수가 Target Files 이하이므로 Compaction을 수행하지 않습니다.")
        return
 
    # 5. Data Column 목록 (Partition Column 제외)
    partition_names = {c.lower() for c in partition_columns}
    data_columns = [
        c for c in spark.table(table_name).columns if c.lower() not in partition_names
    ]
 
    # 6. 대상 Partition 읽기
    where_sql = make_where_sql(partition_spec)
    source_df = spark.sql(f"SELECT * FROM {table_name} WHERE {where_sql}")
    source_count = source_df.count()
    print(f"Source Rows    : {source_count}")
    if source_count == 0:
        print("Partition 데이터가 없습니다.")
        return
 
    # 7. Spark Partition 조절
    if merge_mode == "repartition":
        merged_df = source_df.repartition(target_files)
    else:
        merged_df = source_df.coalesce(target_files)
 
    # 8. Staging ORC Write
    run_id = datetime.now().strftime("%Y%m%d_%H%M%S") + "_" + uuid.uuid4().hex[:8]
    staging_path = (
        f"{staging_base.rstrip('/')}/{table_name.replace('.', '_')}/{run_id}"
    )
    print(f"Staging Path   : {staging_path}")
    merged_df.write.mode("overwrite").format("orc").save(staging_path)
 
    # 9. Staging 검증 — 실패 시 원본 Partition은 손대지 않는다
    staged_df = spark.read.format("orc").load(staging_path)
    staged_count = staged_df.count()
    print(f"Staging Rows   : {staged_count}")
    if source_count != staged_count:
        raise RuntimeError(
            "Source와 Staging Row Count가 일치하지 않습니다. "
            "원본 Partition은 변경하지 않습니다."
        )
 
    # 10. Staging → 원본 Partition INSERT OVERWRITE
    # Staging을 다시 읽으면 Spark가 큰 ORC 파일을
    # spark.sql.files.maxPartitionBytes(기본 128MB) 단위로 쪼개 읽으므로
    # 그대로 INSERT하면 파일 수가 Target Files의 몇 배가 된다.
    # 다시 coalesce하여 최종 파일 수를 Target Files로 맞춘다.
    temp_view = "compact_" + uuid.uuid4().hex
    staged_df.coalesce(target_files).createOrReplaceTempView(temp_view)
 
    select_columns = ", ".join(f"`{c}`" for c in data_columns)
    overwrite_sql = f"""
        INSERT OVERWRITE TABLE {table_name}
        PARTITION ({make_partition_sql(partition_spec)})
        SELECT {select_columns}
        FROM {temp_view}
    """
    print(overwrite_sql)
    spark.sql(overwrite_sql)
 
    # 11. 최종 Row Count 검증 — 실패 시 Staging은 복구용으로 남는다
    target_count = spark.sql(
        f"SELECT COUNT(*) AS cnt FROM {table_name} WHERE {where_sql}"
    ).first()["cnt"]
    print(f"Before Rows    : {source_count}")
    print(f"After Rows     : {target_count}")
    if source_count != target_count:
        raise RuntimeError("Compaction 이후 Row Count가 일치하지 않습니다.")
 
    # 12. 결과 확인 및 정리
    after = get_hdfs_statistics(partition_location)
    print(f"Files          : {current_files} → {after['files']}")
    print(f"Size           : {human_size(partition_size)} → {human_size(after['bytes'])}")
 
    spark.catalog.dropTempView(temp_view)
    if not keep_staging:
        delete_hdfs_path(staging_path)
    else:
        print(f"Staging Directory 유지: {staging_path}")
 
    print("Compaction SUCCESS")
 
 
# ------------------------------------------------------------
# Command Line
# ------------------------------------------------------------
 
def main():
    parser = argparse.ArgumentParser(description="Hive ORC Partition Small File Compaction")
    parser.add_argument("--table", required=True, help="Hive table: db.table")
    parser.add_argument("--partition", action="append", required=True,
                        help="예: --partition dt=2026-10-01 (복수 지정 가능)")
    parser.add_argument("--staging-base", default="hdfs:///tmp/hive_orc_compaction")
    parser.add_argument("--target-file-size-mb", type=int, default=512)
    parser.add_argument("--min-file-count", type=int, default=100)
    parser.add_argument("--target-files", type=int, default=None,
                        help="자동 계산 대신 파일 개수 직접 지정")
    parser.add_argument("--merge-mode", choices=["coalesce", "repartition"], default="coalesce")
    parser.add_argument("--keep-staging", action="store_true")
    args = parser.parse_args()
 
    compact_partition(
        table_name=args.table,
        partition_spec=parse_partition_args(args.partition),
        staging_base=args.staging_base,
        target_file_size_mb=args.target_file_size_mb,
        min_file_count=args.min_file_count,
        target_files_override=args.target_files,
        merge_mode=args.merge_mode,
        keep_staging=args.keep_staging,
    )
 
 
if __name__ == "__main__":
    main()

# 기본: 목표 512 MB, 최소 파일 수 100, coalesce
spark-submit \
  --master yarn \
  --deploy-mode cluster \
  hive_orc_compaction.py \
  --table dw.sales \
  --partition dt=2026-10-01

spark-submit hive_orc_compaction.py --table dw.sales --partition dt=2026-10-01 --target-file-size-mb 256
spark-submit hive_orc_compaction.py --table dw.sales --partition dt=2026-10-01 --target-files 8

spark-submit hive_orc_compaction.py \
  --table dw.sales \
  --partition year=2026 \
  --partition month=10 \
  --partition day=01

Table          : dw.sales
Partition      : dt=2026-10-01
Partition Size : 9.7 GB
ORC Files      : 1,243
Target Size    : 512 MB  →  Target Files ≈ 20

REFRESH dw.sales PARTITION (dt='2026-10-01');

staged_df = spark.read.format("orc").load(
    "hdfs:///tmp/hive_orc_compaction/dw_sales/<run_id>"
)
staged_df.createOrReplaceTempView("compact_recovery")
 
spark.sql("""
    INSERT OVERWRITE TABLE dw.sales
    PARTITION (dt='2026-10-01')
    SELECT id, customer_id, amount
    FROM compact_recovery
""")

ANALYZE TABLE dw.sales PARTITION (dt='2026-10-01') COMPUTE STATISTICS;

원본 Partition → Spark Read → 파일 수 축소 → Staging ORC
  → 건수 검증 → INSERT OVERWRITE PARTITION → 건수 재검증 → Staging 삭제