1. Apache Flink란?
Apache Flink는 상태 기반의 스트림 처리를 위한 오픈소스 분산 처리 엔진입니다. 배치 처리를 스트림 처리의 특별한 경우로 간주하는 유니파이드 데이터 처리 아키텍처를 지향합니다.
실시간 스트림 처리
데이터가 생성되는 즉시 처리하여 밀리초 단위의 저지연 분석이 가능합니다. 이벤트 시간(event time)과 워터마크(watermark)를 통한 정확한 시간 기반 연산을 지원합니다.
상태 관리
분산 상태(state)를 내장 저장소에 안전하게 유지하며, 체크포인트(checkpoint)를 통한 장애 복구와 상태 일관성을 보장합니다. 키별 상태(keyed state)와 연산자별 상태(operator state)를 제공합니다.
정확히 한 번 처리
Chandy-Lamport 알고리즘 기반의 분산 스냅샷을 통해 Exactly-Once 처리 의미론을 보장합니다. 네트워크, 노드 장애 시에도 데이터 손실이나 중복 처리 없이 정확한 결과를 제공합니다.
2. 왜 써야 하는가?
기존 배치 중심 시스템과의 비교를 통해 Flink의 장점을 확인할 수 있습니다.
| 시스템 | 처리 모델 | 지연 시간 | 상태 관리 | 주요 사용 사례 |
|---|---|---|---|---|
| Apache Flink | 스트림 우선 (유니파이드) | 밀리초 ~ 초 | 내장 상태, 체크포인트 | 실시간 대시보드, 이상 감지, CEP, 실시간 ETL |
| Apache Spark | 마이크로 배치 | 초 ~ 분 | 외부 저장소 의존 (RDD) | 배치 분석, 머신러닝, 스트리밍(구조적) |
| Apache Storm | 레코드 단위 | 밀리초 | 외부 상태 저장소 필요 | 실시간 처리 (At-Least-Once) |
| Kafka Streams | 스트림 처리 | 밀리초 | 로컬 상태 저장소 | 카프카 기반 애플리케이션, 단일 앱 처리 |
핵심 선택 기준
• 저지연 실시간 처리가 필수적이고, Exactly-Once 보장이 필요한 경우
• 복잡한 이벤트 시간 기반 윈도우 연산이 필요한 경우
• 대규모 상태(state)를 효율적으로 관리해야 하는 경우
• 배치와 스트림을 통합된 API로 처리하고 싶은 경우
3. 핵심 개념
Flink 애플리케이션은 데이터 소스(Source)에서 싱크(Sink)로 흐르는 데이터 흐름 그래프로 구성됩니다.
DataStream API
무한 데이터 스트림을 처리하기 위한 핵심 API. 변환(Transformation), 조인(Join), 윈도우(Window), 상태(State) 연산을 제공합니다.
이벤트 시간 vs 처리 시간
이벤트 시간: 데이터가 실제 발생한 시간 (데이터에 포함)
처리 시간: 시스템이 데이터를 처리하는 시간
지연/무순서 데이터 처리를 위해 이벤트 시간을 사용해야 정확한 분석 가능.
워터마크(Watermark)
이벤트 시간 진행을 추적하는 메커니즘. "지금까지의 데이터에서 t 시간 이전의 데이터는 모두 도착했다"는 신호를 발생시켜 윈도우가 결과를 출력할 시점을 결정합니다.
체크포인트(Checkpoint)
분산 상태의 일관된 스냅샷. JobManager가 주기적으로 조정하여 모든 작업자의 상태를 저장소에 저장합니다. 장애 발생 시 마지막 체크포인트에서 복구합니다.
4. 퀵스타트
Python API (PyFlink)를 사용한 첫 번째 Flink 애플리케이션 실행하기.
Step 1: 환경 설치
# Python 3.8+ 가 설치되어 있다고 가정
# PyFlink 설치
pip install apache-flink
# 로컬 테스트를 위한 추가 패키지
pip install apache-flink-libraries # Kafka, Hive 등 커넥터
# 버전 확인
python -c "import pyflink; print(pyflink.__version__)"Step 2: 첫 번째 스트리밍 잡 작성 (word_count.py)
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import SocketTextStreamSource
from pyflink.common.typeinfo import Types
from pyflink.datastream.window import TumblingProcessingTimeWindows
from pyflink.common import Time
def word_count():
# 실행 환경 생성
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1) # 병렬도 설정
# 소켓 소스에서 텍스트 스트림 읽기 (로컬 테스트용)
# 터미널에서 `nc -lk 9999` 실행 후 텍스트 입력
text_stream = env.socket_text_stream("localhost", 9999)
# 변환: 단어 분리 → 필터링 → 키별 그룹화 → 윈도우 집계
counts = (
text_stream
.flat_map(lambda line: line.split(" "), output_type=Types.STRING())
.filter(lambda word: len(word) > 0)
.map(lambda word: (word, 1), output_type=Types.TUPLE([Types.STRING(), Types.INT()]))
.key_by(lambda pair: pair[0]) # 단어별 키 지정
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.reduce(lambda a, b: (a[0], a[1] + b[1]))
)
# 결과 출력 (표준 출력)
counts.print()
# 잡 실행
env.execute("Socket Window WordCount")
if __name__ == "__main__":
word_count()Step 3: 애플리케이션 실행
# 터미널 1: Netcat 서버 실행 (데이터 소스)
nc -lk 9999
# 터미널 2: Flink 애플리케이션 실행
python word_count.py
# 터미널 1에서 텍스트 입력 (예: hello world hello flink)
# 터미널 2에서 5초마다 집계 결과 확인
# 출력 예: (hello, 2), (world, 1), (flink, 1)로컬 개발 환경 구성 팁
• PyCharm/VSCode에서 PyFlink 디버깅을 위해 env.set_parallelism(1) 설정 권장
• 실제 운영 환경에서는 Kafka, Kinesis 등의 메시지 큐를 소스/싱크로 사용
• Docker로 Flink 클러스터를 로컬에 띄워 테스트: docker run -p 8081:8081 flink:latest
5. 핵심 설정 레퍼런스
프로덕션 환경에서 중요한 Flink 설정 값들입니다.
| 설정 키 | 기본값 | 설명 | 권장 값 (프로덕션) |
|---|---|---|---|
taskmanager.numberOfTaskSlots |
1 | 각 TaskManager의 태스크 슬롯 수 (CPU 코어당 1개) | 물리적 코어 수 |
parallelism.default |
1 | 기본 병렬도 (애플리케이션 전체) | TaskManager 수 × taskSlots |
execution.checkpointing.interval |
없음 | 체크포인트 간격 (ms) | 1m (60000) |
state.backend |
없음 | 상태 백엔드 유형 | RocksDBStateBackend |
state.checkpoints.dir |
없음 | 체크포인트 저장 디렉토리 | hdfs:///flink/checkpoints |
execution.checkpointing.timeout |
10m | 체크포인트 완료 타임아웃 | 5m |
execution.checkpointing.min-pause |
0 | 체크포인트 사이 최소 간격 | 30s |
restart-strategy |
없음 | 실패 시 재시작 전략 | fixed-delay (3 attempts, 10s delay) |
설정 적용 방법
1. flink-conf.yaml: 클러스터 전체 설정
2. 코드 내 설정: env.set_restart_strategy()
3. 커맨드라인: ./bin/flink run -Dstate.backend=rocksdb ...
4. 파이썬 설정: env.get_config().set_integer("parallelism.default", 4)
6. 실무 팁 & 트러블슈팅
⚠️ 자주 겪는 문제 1: 백프레셔(Backpressure)
증상: 체크포인트 시간 초과, 처리 지연 증가, TaskManager 메모리 부족
원인: 데이터 생성 속도 > 처리 속도, 데이터 스큐, 비효율적 연산자
해결:
1. 병렬도 증가 (set_parallelism())
2. 상태 백엔드를 RocksDB로 변경 (메모리 → 디스크 오프로드)
3. 윈도우 크기 조정 또는 데이터 스큐 처리 로직 추가
4. Web UI의 백프레셔 모니터링 탭 확인
💡 실무 팁 1: 상태 크기 관리
• TTL(Time-To-Live) 설정: 오래된 상태 자동 삭제
from pyflink.common import Time
state_ttl_config = StateTtlConfig.new_builder(Time.days(7)).build()• 체크포인트 간격 조정: 너무 짧으면 오버헤드, 너무 길면 복구 시간 증가
⚠️ 자주 겪는 문제 2: 체크포인트 실패
증상: Checkpoint expired before completing 에러
원인: 네트워크 지연, 저장소 성능 문제, 백프레셔
해결:
1. execution.checkpointing.timeout 값 증가 (10m → 15m)
2. 체크포인트 저장소 성능 확인 (로컬 HDD → SSD 또는 분산 저장소)
3. 최소 간격 설정: execution.checkpointing.min-pause (체크포인트 간 충분한 간격)
💡 실무 팁 2: 이벤트 시간 처리
• 지연 데이터 처리를 위해 워터마크 전략 설정
# 허용 가능한 지연 시간 설정 (예: 5초)
watermark_strategy = WatermarkStrategy\
.for_bounded_out_of_orderness(Duration.of_seconds(5))\
.with_timestamp_assigner(...)• 워터마크 생성 주기 조정: 너무 짧으면 오버헤드, 너무 길면 지연 증가
💡 실무 팁 3: 모니터링 & 디버깅
• Flink Web UI (기본 8081 포트): 실행 계획, 메트릭, 로그 확인
• 메트릭 리포터: Prometheus + Grafana 연동
• 로그 레벨 조정: log4j.properties에서 logger.jobmanager.name = org.apache.flink.runtime.jobmanager.JobManager
• 로컬 테스트: env.execute("job_name") 대신 env.execute_async()로 비동기 실행 가능