Flink는 Confluent Cloud가 제공하는 Managed 버전만 잠깐 사용해본 게 전부였는데 지금 회사에서는 직접 K8s 위에 설치해서 운영 중이다. 운영에 필요한 지식을 공부할 겸 정리하는 포스팅
Flink 클러스터 아키텍처
JobManager
JobManager는 크게 세 가지 역할로 나뉜다.
•
Dispatcher: Job 제출을 받아 JobMaster를 기동시키는 진입점
•
ResourceManager: TaskManager의 리소스(Slot)를 관리하고 필요 시 K8s에 TaskManager Pod를 요청/반환
•
JobMaster: 개별 Job의 실행 계획(DAG) 스케줄링, 체크포인트 코디네이션, 장애 복구 담당
K8s 위에서는 이 세 역할이 하나의 JobManager Pod 안에서 함께 동작한다. Pod 하나에서 세 컴포넌트가 별도 스레드/모듈로 돌아간다고 이해하면 된다.
TaskManager
실제 연산(operator)이 실행되는 워커
•
각 TaskManager는 설정된 수만큼의 Task Slot을 가짐(taskmanager.numberOfTaskSlots)
•
Slot 수는 해당 TaskManager Pod가 동시에 처리할 수 있는 subtask 개수를 결정
•
Job의 병렬성(parallelism)이 크면 필요한 Slot 총량이 늘고 이는 곧 필요한 TaskManager Pod 수로 이어진다
예를 들어 taskmanager.numberOfTaskSlots: 2로 설정하고 Job의 parallelism이 10이면 최소 5개의 TaskManager Pod가 필요하다는 계산이 나온다. K8s 관점에서 Parallelism 설정은 Pod를 몇 개 띄울 것인지 정하는 리소스 설계 문제다.
상태 백엔드와 체크/세이브포인트
먼저 State는 크게 두 종류로 이루어진다.
Keyed State: keyBy()로 나뉜 key별로 유지되는 실제 연산 결과값
Operator State: key와 무관하게 operator 인스턴스 단위로 유지되는 데이터로 Operator State는 Flink의 exactly-once 보장의 핵심 축.
소스별로 어디까지 읽었는지 Operator State에 저장하고 이를 Keyed State와 체크포인트에 원자적으로 묶기 때문에 유실을 방지한다.
이렇게 구성된 State는 TaskManager에서 State Backend에 유지된다.
•
HashMapStateBackend (Heap): State를 TaskManager Pod의 JVM 힙 메모리에 유지한다. 빠르지만 State가 커지면 메모리 부족 위험이 있고 Pod 메모리 크기가 곧 State 크기의 한계가 된다
•
RocksDB (EmbeddedRocksDBStateBackend): State를 TaskManager Pod의 로컬 디스크에 유지한다. 메모리보다 훨씬 큰 State를 다룰 수 있지만 디스크 I/O 때문에 Heap보다 느리다
두 방식 모두 State는 TaskManager Pod 로컬에 저장되므로 Pod가 죽으면 그대로 사라진다. 이를 대비해 Flink는 주기적으로 State를 외부 영구 스토리지(S3 등)에 스냅샷 떠서 백업하는데 이 스냅샷이 Checkpoint와 Savepoint다.
Checkpoint | Savepoint | |
트리거 주체 | Flink가 자동/주기적으로 | 사람 또는 Operator가 명시적으로 |
목적 | 장애 발생 시 자동 복구 | 의도적인 배포/마이그레이션/스케일링 |
보관 | 최근 N개만 자동 유지, 오래된 건 정리 | 명시적으로 지우기 전까지 계속 유지 |
빈도 | 계속 (예: 10초마다) | 필요할 때 1번 |
복구 흐름은 다음과 같다.
1.
TaskManager Pod가 죽고 새로 뜨면 로컬 State는 비어있는 상태로 시작
2.
JobManager는 가장 최근 Checkpoint가 S3의 어느 경로에 있는지 메타데이터를 HA storage에서 가지고 있음
3.
새 TaskManager는 S3에서 해당 Checkpoint를 읽어와 로컬 State를 복원
Job의 배포 모드
Flink Kubernetes Operator는 Application Mode와 Session Mode 두 가지 배포 방식을 지원하며 각각 다른 K8s 리소스 타입으로 관리된다.
Application Mode
Job마다 자체 클러스터를 갖는 방식이다. Application 모드에서는 각 Job이 자신만의 클러스터인 JobManager를 가지고 실행되며 Job이 끝나면 해당 클러스터는 종료된다. K8s 리소스로는 FlinkDeployment를 사용한다.
•
Job 코드(JAR 또는 SQL 실행 로직)가 이미지에 함께 빌드되어 Job 하나 = 클러스터 하나 = 배포 단위 하나
•
Job 간 리소스/장애 격리가 완전함
•
Job 수만큼 JobManager Pod가 늘어나므로 자잘한 Job이 많으면 오버헤드가 커질 수 있음
•
CI/CD 파이프라인에 태우기 좋은 구조 (이미지 빌드 → 배포 → 종료가 명확)
Session Mode
장기 실행되는 하나의 클러스터에 여러 Job을 제출하는 방식이다. Session 모드는 여러 Job이 배포되는 장기 실행 Flink 클러스터를 사용하며 FlinkSessionJob 리소스 타입을 통해 관리된다. K8s 리소스로는 FlinkDeployment(클러스터 정의) + FlinkSessionJob(개별 Job)을 함께 사용한다.
•
JobManager/TaskManager를 여러 Job이 공유 → 리소스 효율은 좋음
•
하지만 공유 클러스터 구조라 리소스를 절약할 수 있지만 업그레이드 조율이나 리소스 경합 같은 운영 복잡도가 따라온다.
•
한 Job의 장애나 리소스 폭증이 같은 클러스터의 다른 Job에 영향을 줄 수 있음
•
반대로 Job 시작/제출 자체는 빠름 (이미 떠 있는 클러스터에 붙기만 하면 됨)
비교 표
기준 | Application Mode | Session Mode |
격리성 | 높음 (Job = 클러스터) | 낮음 (클러스터 공유) |
리소스 효율 | 낮음 (Job마다 JobManager) | 높음 (JobManager 공유) |
배포 단위 | FlinkDeployment | FlinkDeployment + FlinkSessionJob |
운영 복잡도 | 상대적으로 단순 | 업그레이드/경합 관리 필요 |
적합한 상황 | Job이 크고 독립적, 배포 빈도가 개별적 | 소규모 Job이 많고 공유해도 무방한 경우 |
Flink Kubernetes Operator
Operator는 K8s의 Custom Resource(CRD)를 감시하면서 Flink 클러스터/Job의 생명주기를 자동으로 관리하는 컨트롤러다.
CRD 구조
FlinkDeployment: Application 배포는 Application Mode에서 단일 Job 배포를 관리하고, Session 배포는 별도의 Job 관리 없이 Flink Session 클러스터 자체를 관리한다.
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: order-aggregation-job
spec:
image: my-registry/order-aggregation:1.2.0 # Job 코드가 빌드된 이미지
flinkVersion: v1_18
serviceAccount: flink
flinkConfiguration:
state.backend: rocksdb
state.checkpoints.dir: s3://flink-checkpoints/order-aggregation
high-availability: kubernetes
high-availability.storageDir: s3://flink-ha/order-aggregation
jobManager:
resource:
memory: "2048m"
cpu: 1
taskManager:
resource:
memory: "4096m"
cpu: 2
job:
jarURI: local:///opt/flink/usrlib/order-aggregation.jar
parallelism: 4
upgradeMode: last-state
YAML
복사
•
FlinkSessionJob: Session 클러스터 위에 제출되는 개별 Job을 정의. 하나의 Session 클러스터에 여러 FlinkSessionJob이 붙을 수 있다.
# 1) 클러스터 자체를 정의 (JM/TM만 뜨고, Job은 없는 상태)
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: shared-session-cluster
spec:
image: flink:1.18
flinkVersion: v1_18
flinkConfiguration:
taskmanager.numberOfTaskSlots: "4"
jobManager:
resource:
memory: "2048m"
cpu: 1
taskManager:
resource:
memory: "4096m"
cpu: 2
# job 필드가 없으면 Session 클러스터로 뜸
---
# 2) 이 클러스터 위에 Job을 하나 제출
apiVersion: flink.apache.org/v1beta1
kind: FlinkSessionJob
metadata:
name: click-stream-job
spec:
deploymentName: shared-session-cluster # 위에서 정의한 클러스터 이름 참조
job:
jarURI: s3://jobs/click-stream.jar
parallelism: 2
upgradeMode: stateless
YAML
복사
Native vs Standalone 배포 방식
Operator는 클러스터를 실제로 어떻게 K8s에 붙일지에 대해 두 가지 모드를 지원한다. 핵심 차이는 TaskManager Pod를 누가 어떻게 만드는지. Flink 프로세스 자신이 K8s API를 직접 호출해서 Pod를 만드는지 아니면 누군가 미리 만들어둔 Pod 위에 얹혀서 도는 것뿐인지의 차이다.
•
Native
◦
Flink 자체에 내장된 K8s 통합 기능을 사용하는 기본 배포 모드
◦
Flink 클러스터가 K8s와 직접 통신하며 TaskManager Pod를 동적으로 할당/해제하는 등 K8s 리소스를 직접 관리 가능
◦
Job의 parallelism이 바뀌면 → ResourceManager가 필요한 Slot을 계산 → 부족하면 K8s API로 TaskManager Pod를 새로 띄우라고 직접 요청
•
Standalone
◦
K8s를 단순히 오케스트레이션 플랫폼으로만 사용하는 방식으로 Flink는 자신이 K8s 위에서 실행되고 있다는 사실을 인지하지 못함
◦
TaskManager Pod들은 일반 K8s Deployment/ReplicaSet처럼 미리 정해진 개수(replica)로 고정되어서 뜬다
◦
Job의 parallelism이 늘어도 Flink가 알아서 Pod를 늘리지 않음 → 사람이 직접 kubectl scale로 replica 수를 늘려줘야 함
Reconcile 루프
Operator는 사용자가 제출한 FlinkDeployment/FlinkSessionJob의 spec(desired state)과 실제 클러스터 상태를 지속적으로 비교하며, 차이가 있으면 이를 맞추는 방향으로 동작한다. 예를 들어 이미지 태그를 바꿔서 spec을 업데이트하면, Operator가 이를 감지해 롤링 업그레이드를 트리거하는 식이다.
Job 생명주기 관리
Job을 재시작/업그레이드할 때 상태를 어떻게 다룰지는 upgradeMode로 제어
•
stateless
◦
기존 Job을 그냥 죽이고 새 Job을 처음부터 처음부터 다시 시작
◦
배포 전까지 집계값은 다 날아가기 때문에 상태가 중요하지 않은 단순 필터링 Job에 사용
•
savepoint
◦
업그레이드 전 Operator가 명시적으로 Savepoint 생성
◦
기존 Job을 내리고 새 Job을 그 Savepoint부터 이어서 시작
◦
상태가 유지되지만 Savepoint를 새로 찍는 과정이 있어서 배포 시간이 조금 더 걸림
•
last-state
◦
Savepoint를 새로 찍지 않고 이미 주기적으로 쌓이고 있던 가장 최근 체크포인트를 그대로 이어받음
◦
새로 찍는 과정이 없어 savepoint 모드보다 빠름
◦
HA(High Availability) 설정이 켜져 있어야 안정적으로 동작