라떼쉬폰의 코드 베이커리

PostgreSQL 데이터를 Parquet로 이관하는 Cold Storage 파이프라인

·

처음에는 애플리케이션에서 SELECT 결과를 읽어 Parquet 파일로 쓰려고 했습니다. 데이터가 적을 때는 잘 동작했습니다.

범위가 커지자 모든 직렬화와 전송이 애플리케이션을 통과했습니다. Migration Job의 메모리 사용량과 실행 시간이 데이터량에 따라 크게 흔들렸습니다. Cold Storage의 목적은 단순히 파일 하나를 만드는 것이 아니었습니다.

이관이 끝난 범위는 온라인 PostgreSQL에서 정리해야 했고, DuckDB로 다시 읽어도 같은 데이터라는 증거도 필요했습니다. 중복 Job, 부분 파일, manifest 실패, source partition 누락을 구분할 수 있어야 했습니다. 그래서 PostgreSQL 조회와 Parquet COPY는 DuckDB가 맡고, 애플리케이션은 범위와 상태만 관리하도록 나눴습니다.

시간 window와 dataset 단위로 나눈 이관

이벤트, 세그먼트, 결과 항목, statistics bucket은 서로 다른 테이블이지만 같은 시간 생명주기를 가집니다. Migration Job은 먼저 [startAt, endAt) window를 계획합니다. 그 안에서 dataset을 순서대로 이관합니다.

data class MigrationWindow(
    val startAt: Instant,
    val endAt: Instant,
) {
    init {
        require(startAt < endAt)
    }
}

data class DatasetSpec(
    val dataset: ColdDataset,
    val tableName: String,
    val timeColumn: String,
    val usesTenDayPartition: Boolean,
)

모든 범위는 반개구간으로 통일했습니다. 인접 window에서도 경계 행은 중복되지 않습니다. Hot·Cold 3계층 조회 구조의 라우터도 같은 규칙을 씁니다.

dataset마다 시간 컬럼이 다를 수 있습니다. 이벤트 계열은 occurred_at, 통계 bucket은 bucket_start를 사용합니다. 시간 규칙은 spec 한 곳에 모았습니다.

이렇게 하면 count와 COPY, manifest가 모두 같은 조건을 사용합니다.

Source partition 존재 여부 확인

Hot 테이블이 10일 단위 partition이면 이관 window에서 읽을 실제 자식 테이블 이름을 계산할 수 있습니다. 하지만 신규 dataset은 과거 backfill이 없어 해당 partition 자체가 없을 수 있습니다. 모든 누락을 오류로 처리하면 데이터가 없는 정상 범위도 Job을 막습니다.

SELECT to_regclass('public.event_record_p20260801') IS NOT NULL;

dataset spec에 missing source partition을 빈 범위로 허용할지 명시했습니다. 원래 반드시 존재해야 하는 핵심 dataset의 partition이 없다면 즉시 실패합니다. 신규 통계처럼 과거 데이터가 없다는 것이 정상인 dataset만 건너뜁니다.

source table 이름은 검증된 spec과 날짜로 만듭니다. 그 뒤 SQL identifier 규칙을 검사합니다. 테이블 이름과 파일 경로는 직접 붙이지 않았습니다.

SQL injection과 경로 이탈을 막기 위해서입니다.

COPY 전 원본 행 수 기록

Parquet 생성이 성공해도 내용이 비어 있거나 일부만 들어갈 수 있습니다. 먼저 PostgreSQL에서 동일 조건의 row count를 구해 manifest의 기대값으로 저장했습니다.

SELECT count(*)
FROM event_record_p20260801
WHERE occurred_at >= :start_at
  AND occurred_at < :end_at;

count가 0이면 파일을 만들지 않고 빈 범위로 처리할 수 있습니다. 다만 빈 dataset이 전체 window의 coverage 진행을 막아야 하는지는 별도 정책입니다. 0건이 정상인 dataset도 있습니다.

부모 이벤트가 있어도 자식은 0건일 수 있습니다. 대규모 범위에서 사전 count(*)는 추가 스캔 비용입니다. 파일을 만든 뒤 Parquet metadata의 row count와 원본을 비교하는 방식도 가능합니다.

Hot 데이터를 지울 검증 근거가 필요했습니다. window도 partition 크기로 제한됩니다. 따라서 명시적인 count를 선택했습니다.

애플리케이션 직렬화 대신 DuckDB를 선택한 이유

Job은 전용 DuckDB connection에 postgres extension을 로드하고 Hot PostgreSQL을 attach합니다. 그다음 검증된 source table과 시간 조건으로 COPY를 실행합니다.

INSTALL postgres;
LOAD postgres;

ATTACH 'host=... port=5432 dbname=... user=... password=...'
AS hot (TYPE POSTGRES);

COPY (
    SELECT *
    FROM hot.event_record_p20260801
    WHERE occurred_at >= TIMESTAMPTZ '2026-08-01T00:00:00Z'
      AND occurred_at <  TIMESTAMPTZ '2026-08-11T00:00:00Z'
) TO '/cold/dataset=event_record/start=20260801000000/run=.../data.parquet'
  (FORMAT PARQUET);

애플리케이션은 각 row를 Kotlin 객체로 만들지 않습니다. DuckDB가 PostgreSQL wire protocol에서 읽은 데이터를 columnar file로 기록합니다. JVM 메모리는 데이터량보다 query와 파일 buffer에 가까운 수준으로 유지됩니다.

JDBC row streaming과 직접 writer를 조합하면 메모리를 가장 세밀하게 통제할 수 있지만 구현 책임이 커집니다. 전체 조회 후 객체로 변환하는 방식은 쉽지만 데이터량에 따라 메모리가 급증합니다. PostgreSQL 서버 측 export는 애플리케이션 부하가 낮아도 Parquet 지원에 제약이 있습니다. 이 작업에서는 메모리 사용이 일정하고 SQL로 이관 과정을 표현할 수 있는 DuckDB postgres attach와 COPY를 선택했습니다.

DuckDB 방식에서도 database connection과 query concurrency를 제한해야 합니다. 여러 Migration Job이 동시에 큰 partition을 읽으면 Hot PostgreSQL I/O를 압박할 수 있습니다.

append-only 파일 경로

같은 범위를 재실행할 때 기존 파일을 덮어쓰면 조회 중인 reader가 부분 파일을 볼 수 있습니다. dataset, 시작, 끝, run UUID를 경로에 포함해 매 실행이 새 파일을 만들도록 했습니다.

/cold/
  dataset=event_record/
    start=20260801000000/
      end=20260811000000/
        run=6a2f.../
          data.parquet

manifest가 새 run을 완료 상태로 전환한 뒤 이전 run을 superseded로 표시합니다. reader는 완료된 최신 manifest만 사용합니다. 실패한 run의 파일은 조회 대상이 아니며 별도 cleanup 정책으로 제거합니다.

base path를 toAbsolutePath().normalize()한 뒤 생성 경로가 그 아래인지 검증했습니다. Parquet path를 DuckDB SQL에 넣을 때 single quote를 허용하지 않거나 escape 규칙을 고정해 파일 경로가 SQL 조각으로 변하지 않게 했습니다.

Manifest를 쓰기 전·후 상태로 나눴다

파일 작업과 PostgreSQL manifest 갱신은 하나의 트랜잭션이 아닙니다. 상태를 한 번에 SUCCEEDED로 insert하면 프로세스 종료 지점을 설명할 수 없습니다.

  1. 기대 row count와 최종 파일 경로를 가진 WRITING manifest를 만듭니다.
  2. Parquet COPY를 실행합니다.
  3. 파일 존재, 크기, row count와 schema를 검증합니다.
  4. manifest를 SUCCEEDED로 바꾸고 같은 범위의 이전 성공본을 supersede합니다.
  5. 예외가 나면 manifest를 FAILED로 표시합니다.
val manifestId = manifestRepository.createWriting(
    dataset = spec.dataset,
    rangeStart = startAt,
    rangeEnd = endAt,
    filePath = path.toString(),
    rowCount = sourceCount,
    schemaVersion = schemaVersion,
)

try {
    duckDb.sql(copySql).update()
    verifyParquet(path, sourceCount)
    manifestRepository.completeAndSupersedePrevious(manifestId, spec.dataset, startAt, endAt)
} catch (e: Exception) {
    manifestRepository.markFailed(manifestId)
    throw e
}

실패 상태 갱신 자체도 실패할 수 있으므로 다음 Job 시작 시 오래된 WRITING manifest와 orphan 파일을 복구하는 절차가 필요합니다.

모든 dataset 뒤의 coverage 확인

이벤트 파일만 성공하고 결과 항목 파일이 실패했는데 Cold coverage를 먼저 앞으로 옮기면 조회 라우터가 불완전한 범위를 Cold로 보냅니다. 각 window의 모든 필수 dataset 처리가 끝난 뒤에만 dataset별 연속 coverage를 갱신했습니다. coverage는 성공 manifest의 최대 end가 아닙니다.

[1, 11), [21, 31) 파일이 있어도 [11, 21)이 비면 연속 범위는 11에서 멈춰야 합니다. manifest repository가 현재 coverage 끝에서 바로 이어지는 성공 범위만 advance하도록 했습니다. schema version도 coverage 조건에 포함했습니다.

파일 스키마가 바뀌었을 때 reader가 구버전과 신버전을 모두 읽을 수 있는지, 또는 version별로 경로를 나눌지 정해야 합니다.

중복 실행을 막는 advisory lock

외부 스케줄러 중복 실행이나 수동 재실행으로 두 Job이 같은 범위를 처리할 수 있습니다. 파일 경로는 UUID로 충돌하지 않더라도 Hot I/O가 두 배로 발생하고 manifest 경쟁이 생깁니다.

SELECT pg_try_advisory_lock(:migration_lock_key);

Job 시작 시 PostgreSQL advisory lock을 시도하고 얻지 못하면 다른 migration이 실행 중인 것으로 보고 종료합니다. session-level lock이라면 같은 JDBC connection을 작업 끝까지 유지하고 finally에서 해제해야 합니다. connection pool이 다른 세션으로 SQL을 보내지 않도록 lifecycle을 확인하는 것도 중요합니다.

전역 lock은 구현이 단순하지만 서로 다른 dataset과 window도 직렬화합니다. 이관 시간이 길어져 병렬성이 필요해지면 dataset과 window로 lock key를 세분화하되 Hot DB의 허용 I/O를 넘지 않게 별도 concurrency 제한을 둬야 합니다.

Hot 삭제는 이관 Job과 분리했다

Parquet 파일과 manifest가 성공했다고 즉시 source partition을 drop하지 않았습니다. 잘못된 reader 배포, 파일 접근 권한, schema 호환성 문제를 발견할 시간을 두는 것이 안전합니다. Hot 정리 조건은 다음과 같이 더 보수적으로 잡았습니다.

  • 모든 필수 dataset의 성공 manifest가 존재합니다.
  • source count와 Parquet count가 일치합니다.
  • 연속 coverage가 해당 window를 포함합니다.
  • 보존 유예 기간이 지났습니다.
  • reader가 Cold 범위 조회 smoke test를 통과했습니다.

정리는 partition detach 후 일정 기간 보관하고 최종 drop하는 두 단계로 만들 수 있습니다. 삭제는 복구 비용이 크므로 Migration 성공의 자동 부수 효과로 두지 않는 편이 낫습니다.

운영에서 확인한 항목

  • dataset·window별 count, COPY, 검증, manifest 소요 시간.
  • 임시·최종 파일 크기와 사용 가능한 디스크 공간.
  • FAILED 또는 오래된 WRITING manifest.
  • orphan Parquet 파일과 superseded 파일 수.
  • Hot source count와 Parquet row count 차이.
  • coverage gap과 schema version 분포.
  • advisory lock 획득 실패와 전체 Job 실행 시간.
  • DuckDB query concurrency와 PostgreSQL read I/O.

Job 로그는 하나의 window 안에서도 count, COPY, manifest, coverage 단계를 분리했습니다. 마지막 로그가 어디에서 끊겼는지만으로 재시작 위치와 잔여 상태를 추정할 수 있게 하기 위해서입니다.

삭제 가능한 Cold 데이터의 증거

PostgreSQL 데이터를 Parquet로 옮기는 핵심은 빠른 파일 생성이 아니라 Hot 데이터를 지워도 된다는 증거를 만드는 일입니다. DuckDB의 postgres attach와 COPY를 사용해 대량 데이터를 JVM 객체로 만들지 않고 이동했지만, 실제 안전성은 manifest와 coverage에서 나왔습니다. 시간 window와 dataset 계약, append-only 파일 경로, WRITING에서 SUCCEEDED로의 상태 전이, 연속 coverage, advisory lock, 삭제 유예가 함께 있어야 합니다.

Cold Storage는 단순한 백업 파일이 아니라 별도의 읽기 계층입니다. 생성, 검증, 조회, 폐기까지 하나의 생명주기로 설계해야 운영에서 신뢰할 수 있습니다.

댓글

답글 남기기

이메일 주소는 공개되지 않습니다. 필수 필드는 *로 표시됩니다