Savepoint

상위: Flink

요약

Flink의 SavePoint 기능을 다룹니다. Checkpoint와 동일한 메커니즘을 사용하지만 사용자가 수동으로 생성하고 관리하는 저장점으로, 작업 종료 후에도 복원이 가능한 장기 보존용 상태 저장 방법을 학습합니다.

SavePoint

• 게임에서의 '저장'과 같은 개념
• Checkpoint와 동일한 매커니즘 사용
• 사용자가 직접 생성 & 삭제해야 함
• 작업 종료 후에도 복원 가능

SavePoint 사용법

• job 실행 중에만 SavePoint 생성 가능

• CMD에 SavePoint 명령 전달 필요

• 결과 파일 경로 예시:

  /tmp/flink-savepoints/savepoint-{JOB-ID}

SavePoint 생성 예제

# 체크포인트 활성화
env.enable_checkpointing(5000)
env.get_checkpoint_config().set_checkpoint_storage(
    FileSystemCheckpointStorage(CHECKPOINT_PATH)
)

# 데이터 로드 및 처리
csv_path = "../data/data.csv"
df = pd.read_csv(csv_path)
transactions = df[['transaction_id', 'amount']].dropna().values.tolist()
transaction_stream = env.from_collection(transactions)

# 연산 적용
processed_stream = transaction_stream.map(TransactionProcessor())
processed_stream.print()

# 비동기 실행
job_client = env.execute_async("Savepoint Debugging Example")
job_id = job_client.get_job_id()
time.sleep(5)

# SavePoint 생성 명령어 실행
savepoint_command = f"$FLINK_HOME/bin/flink savepoint {job_id} {SAVEPOINT_PATH}"
os.system(savepoint_command)

SavePoint vs Checkpoint

SavePoint 생성 과정

1. 작업 실행 중 SavePoint 요청

# Flink CLI를 통한 SavePoint 생성
./bin/flink savepoint <job-id> <savepoint-directory>

2. SavePoint 생성 과정

1. Barrier 전송: 모든 소스에 SavePoint Barrier 전송
2. 상태 저장: 각 연산자의 상태를 지정된 디렉토리에 저장
3. 메타데이터 기록: SavePoint 정보를 메타데이터에 기록
4. 완료 알림: JobManager에 SavePoint 완료 알림

3. SavePoint 파일 구조

/tmp/flink-savepoints/savepoint-{job-id}/
├── _metadata
├── shared/
│   └── {operator-id}/
└── taskowned/
    └── {task-id}/

SavePoint 활용 시나리오

1. 버전 업그레이드

# 현재 버전에서 SavePoint 생성
./bin/flink savepoint <job-id> /savepoints/v1.0/

# 새 버전으로 복원
./bin/flink run -s /savepoints/v1.0/ new-version-job.jar

2. A/B 테스팅

# 원본 작업에서 SavePoint 생성
./bin/flink savepoint <original-job-id> /savepoints/baseline/

# 실험 버전으로 복원
./bin/flink run -s /savepoints/baseline/ experimental-job.jar

3. 장기 보존

# 중요한 상태를 SavePoint로 보존
./bin/flink savepoint <job-id> /backup/monthly-savepoint/

SavePoint 복원

1. 동일한 애플리케이션으로 복원

./bin/flink run -s /savepoints/savepoint-{job-id}/ original-job.jar

2. 수정된 애플리케이션으로 복원

./bin/flink run -s /savepoints/savepoint-{job-id}/ modified-job.jar

3. 복원 시 주의사항

• 호환성 확인: SavePoint 생성 시점과 복원 시점의 애플리케이션 호환성
• 상태 스키마: 상태 구조 변경 시 마이그레이션 필요
• 외부 의존성: 외부 시스템과의 연결 상태 확인

SavePoint 관리

1. SavePoint 목록 확인

# SavePoint 디렉토리 확인
ls -la /tmp/flink-savepoints/

2. SavePoint 삭제

# 개별 SavePoint 삭제
rm -rf /tmp/flink-savepoints/savepoint-{job-id}/

# 오래된 SavePoint 정리
find /tmp/flink-savepoints/ -type d -mtime +30 -exec rm -rf {} \;

3. SavePoint 백업

# SavePoint를 다른 위치로 백업
cp -r /tmp/flink-savepoints/savepoint-{job-id}/ /backup/savepoints/

SavePoint 최적화

1. 저장소 선택

• 로컬 파일시스템: 개발/테스트 환경
• HDFS: 대규모 프로덕션 환경
• S3: 클라우드 환경

2. 압축 설정

# SavePoint 압축 활성화
env.get_checkpoint_config().set_checkpoint_storage(
    FileSystemCheckpointStorage("file:///tmp/flink-savepoints", True)
)

3. 정기적 정리

# 오래된 SavePoint 자동 정리
def cleanup_old_savepoints(savepoint_dir, days=30):
    import os
    import time
    
    current_time = time.time()
    cutoff_time = current_time - (days * 24 * 60 * 60)
    
    for item in os.listdir(savepoint_dir):
        item_path = os.path.join(savepoint_dir, item)
        if os.path.isdir(item_path):
            if os.path.getmtime(item_path) < cutoff_time:
                os.rmtree(item_path)
                print(f"Deleted old savepoint: {item}")

관련 개념

• Checkpointing - 자동 체크포인트 메커니즘
• State Management - 상태 관리 기본 개념
• Architecture - Flink 아키텍처 구조