Apache Flink: Stream and Batch Processing in a Single Engine

초록

• Flink: 스트리밍 데이터와 배치 데이터를 처리하는 오픈소스 시스템
• 실시간 분석, 연속적인 데이터 파이프라인, 과거 데이터 처리(batch), 반복 알고리즘(ML, graph analysis)을 전부 파이프라인 형태의 장애 허용 dataflow로 표현하고 실행할 수 있다는 철학
• 겉보기엔 꽤 다른 사용 사례들을 하나의 실행 모델(single execution model) 아래에서 통합

1. INTRODUCTION

• 전통적으로 data-stream processing과 static(batch) data processing은 서로 다른 종류의 애플리케이션
  • programming model, API, 실행 시스템이 다 달랐음
  • streaming: Storm, Infosphere Streams, StreamInsight, Streambase …
  • batch: relational DB, Hadoop 기반 엔진, Spark, Drill …
• 그런데 실제 데이터는 대부분 시간의 흐름에 따라 연속적으로 생성됨
  • web log, application log, sensor, DB의 transaction log
• 기존 시스템은 이걸 무시하고 record를 인위적으로 묶어서 처리

hourly / daily / monthly
→ static dataset(batch)으로 묶고
→ data collection tool + workflow manager + scheduler로 조율

• Lambda Architecture: batch와 stream 시스템을 결합

Streaming fast path → 빠르지만 approximate한 결과
Batch offline path  → 늦지만 accurate한 결과

• 문제
  • batch로 인한 높은 latency
  • 여러 시스템을 연결하고 조율하는 복잡도
  • 같은 business logic을 두 번 구현
  • application code가 time dimension을 명시적으로 다루지 않아서 생기는 부정확성

Flink의 접근

• data-stream processing 자체를 통합 모델로 사용. programming model뿐 아니라 execution engine에도 적용함
• Kafka, Kinesis처럼 임의 시점부터 replay 가능한 durable message queue와 결합하면, 아래 셋은 본질적으로 크게 다르지 않음
  • 최신 event를 real-time으로 처리하기
  • 큰 window 단위로 지속적으로 aggregate하기
  • 수 TB의 historical data를 처리하기
  • 즉 durable stream의 서로 다른 지점에서 처리를 시작하고, 서로 다른 state를 유지할 뿐!
• 유연한 windowing으로 early + approximate와 delayed + accurate 결과를 한 연산에서 같이 계산하므로 stream과 batch 시스템을 따로 조합할 필요가 없음
• time notion: event time, ingestion time, processing time

Batch는 streaming의 특수한 경우

• 그렇다고 batch가 필요 없다는 말은 아님
  • static dataset에 대한 복잡한 query, legacy 구현, 아직 효율적인 streaming algorithm이 없는 분석
• Flink에서 batch program은 유한한 stream을 처리하는 streaming program
  • record의 order와 time은 중요하지 않고, 모든 record가 하나의 거대한 window에 속한다고 볼 수 있음
• 다만 batch를 효율적으로 돌리기 위해 별도 최적화는 제공
  • static dataset용 API, join과 grouping용 자료구조와 알고리즘, 전용 scheduling
• 그래서 streaming runtime 위에서 돌면서도 완전한 batch processor처럼 동작

2. SYSTEM ARCHITECTURE

Libraries
APIs        (DataSet / DataStream)
Core        (distributed streaming dataflow runtime)
Deployment  (local / cluster / cloud)

• 핵심은 distributed dataflow engine
• runtime program은 stateful operator와 operator를 잇는 data stream으로 된 DAG
• 당시 core API 두 개

DataSet API    → finite dataset            → batch
DataStream API → potentially unbounded     → stream

• 두 API 모두 결국 같은 streaming dataflow runtime용 program을 만듦. runtime이 bounded와 unbounded를 모두 추상화하는 공통 기반이 됨

Cluster

Client ──(dataflow graph)──▶ JobManager ──▶ TaskManager(s)

• Client: program code를 dataflow graph로 변환해 JobManager에 제출. 이때 operator 사이 data type을 분석해 serializer 같은 type-specific code도 생성
• JobManager: 분산 실행 조정
  • operator와 stream의 state, progress를 추적
  • scheduling, checkpoint coordination, recovery coordination
  • HA 구성에서는 checkpoint마다 최소 metadata를 fault-tolerant storage에 저장해 두면 standby JobManager가 이어받을 수 있음
• TaskManager: 실제 데이터 처리
  • operator를 실행하고 상태를 JobManager에 보고
  • buffer pool, operator 사이 network connection 관리

⭐ 3. THE COMMON FABRIC: STREAMING DATAFLOWS

• API가 뭐든 모든 프로그램은 같은 representation, Dataflow Graph로 변환됨

3.1. Dataflow Graphs

• DAG는 stateful operator와 data stream으로 구성
• data-parallel 실행
  • operator 하나는 여러 병렬 instance로 나뉘고, 이게 subtask
  • stream도 여러 stream partition으로 나뉨 (producing subtask마다 하나)

Operator = logical processing node
Subtask  = parallel runtime instance of an operator

Logical                 Physical (parallelism = 3)

Source Operator
      ↓
Map Operator      →     Map Operator
      ↓                  ├─ Subtask 0
Window Operator          ├─ Subtask 1
      ↓                  └─ Subtask 2
Sink Operator

• operator 예: filter, hash join, stream window …
• stateful operator가 일반적인 경우이고 stateless는 그 특수한 경우
• stream 전달 패턴: point-to-point, broadcast, repartition, fan-out, merge

3.2. Data Exchange through Intermediate Data Streams

• intermediate data stream: 한 operator가 만든 데이터를 다른 operator가 consume하도록 잇는 logical handle
  • 여기서 logical은 반드시 disk에 materialize된다는 뜻이 아님

1. Pipelined

• producer와 consumer가 동시에 실행. 데이터가 만들어지는 즉시 downstream으로 보냄
• consumer가 느리면 그 영향이 upstream까지 전달됨: backpressure
• batch dataflow에서도 가능하면 pipelined stream을 써서 불필요한 materialization을 피함

2. Blocking

• bounded stream에서만 씀. producer 데이터를 전부 buffering한 뒤 consumer가 사용

Producer → ALL DATA materialized → Consumer

• producer와 consumer를 다른 execution stage로 분리
• memory를 더 쓰고, disk로 spill될 수 있고, backpressure를 전달하지 않음
• sort-merge join처럼 pipeline을 끊어야 하는 연산에 사용

Latency vs Throughput

• data exchange는 buffer 기반. record를 serialize해서 buffer에 넣고

  1. buffer가 가득 찼을 때
  2. timeout이 났을 때

  consumer로 보냄

큰 buffer / 긴 timeout → throughput ↑, latency ↑
짧은 timeout          → latency ↓

Control Events

• stream에는 data record만이 아니라 control event도 흐름
  • Checkpoint Barrier: stream을 pre-checkpoint와 post-checkpoint로 나눔
  • Watermark: stream partition에서 event time이 어디까지 진행됐는지 알림
  • Iteration Barrier: iterative algorithm에서 superstep의 끝을 표시

Ordering

• partition 하나를 consume하는 unary operator는 FIFO 보장
• 여러 partition을 받으면 arrival order대로 merge하므로 repartition이나 broadcast 뒤에는 전체 순서가 보장되지 않음
• 의도적인 설계!
  • map, hash join 같은 대부분의 operator는 global order가 필요 없음
  • event-time window처럼 순서가 중요한 operator는 자기 logic 안에서 out-of-order를 더 효율적으로 처리

⭐ 3.3. Fault Tolerance

• checkpointing과 partial re-execution으로 exactly-once processing consistency를 보장
• 전제: source가 persistent하고 replayable해야 함 (file, Kafka)
• 왜 lineage만으로는 안 되냐면,,,
  • unbounded stream job은 수개월 돌 수 있음. 처음부터 다시 계산하면 수개월치를 replay해야 함
  • 그래서 일정 interval마다 operator state + input stream position을 snapshot함

Asynchronous Barrier Snapshotting (ABS)

• 문제: 여러 parallel operator의 state를 실행을 멈추지 않고 같은 logical time 기준으로 찍어야 함
• barrier = input stream에 끼워 넣는 특별한 control record

snapshot에 포함될 record
        │
     barrier
        │
다음 snapshot에 포함될 record

Barrier Alignment

Input A ──▶
            Operator
Input B ──▶

• input이 여러 개면 모든 input에서 같은 checkpoint의 barrier가 올 때까지 기다림. 이게 alignment
• 다 오면 operator state(window content 등)를 durable storage에 쓰고, barrier를 downstream으로 넘김
• 모든 operator가 snapshot을 등록하면 global snapshot 완성

Chandy–Lamport와의 관계

• 비슷하지만, Flink program은 DAG라서 in-flight record를 checkpoint할 필요가 없음
• alignment로 record의 effect를 operator state에 반영시킨 뒤 state만 저장
• 그래서 저장량을 이론상 최소, 즉 현재 operator state 정도로 줄임

Recovery

• 모든 operator state를 마지막 성공 snapshot으로 되돌리고, source도 그 barrier 시점부터 다시 읽음
• recomputation은 최대 checkpoint N ~ N+1 사이의 record 정도
• 장점
  1. 계산을 멈추지 않고 exactly-once state update가 가능
  2. checkpoint와 window triggering 같은 다른 control message가 분리됨
  3. state storage와도 분리되어 filesystem, DB 등 아무 storage나 쓸 수 있음

Checkpoint ≠ Micro-batch ≠ Window

• 여기서 제일 중요한 것: checkpoint interval과 window interval은 묶여 있지 않다

checkpoint every 10 sec  ≠  window = 10 sec

• 그리고 micro-batch와도 다름

Spark Streaming
  epoch = execution unit

Flink
  record = execution unit
  epoch  = snapshot consistency boundary

• Flink는 record 단위로 계속 흘려보내고, checkpoint는 “어디까지 일관되게 반영됐나"를 긋는 경계일 뿐!!

3.4. Iterative Dataflows

• graph processing, ML에서 iteration이 중요
• 기존 플랫폼: iteration마다 새 job을 제출하거나, running DAG에 node를 추가하거나, feedback edge를 추가함
• Flink: iteration step operator가 내부에 자체 execution graph를 가짐
  • DAG runtime을 유지하면서 iteration head/tail task로 feedback channel을 표현
  • BSP 같은 structured iteration을 위한 coordination도 제공

4. STREAM ANALYTICS ON TOP OF DATAFLOWS

• DataStream은 특정 type의 element로 된 potentially unbounded immutable collection
• runtime이 이미 pipelined transfer, continuous stateful operator, fault tolerance를 주니까 stream processor에 더 필요한 건 사실상

window system + state interface

• runtime 입장에서 window도 특별한 개념이 아니라 stateful operator의 한 구현일 뿐 ← 꽤 중요한 문장

4.1. The Notion of Time

• Event time: event가 실제로 발생한 시간 (e.g. sensor timestamp)
• Processing time: 처리하는 machine의 wall-clock
• 둘 사이에 arbitrary skew가 생길 수 있음

event 발생: 10:00:01
Flink 도착: 10:00:12

Watermark

• event time으로 처리하면서 무한정 기다리지 않으려고 low watermark를 삽입
• watermark t는 timestamp가 t보다 작은 event가 이미 다 들어왔다고 간주한다는 뜻
• source에서 만들어져 downstream으로 전달
  • map, filter → 그대로 전달
  • event-time window → watermark로 먼저 계산하고 전달
  • input이 여러 개면 minimum을 전달 (느린 input에서 과거 event가 올 수 있으니까)

Processing time vs Event time

processing time: latency 낮음 / recovery 후 replay하면 결과가 달라질 수 있음
event time:      semantic 안정적 / 도착 지연만큼 latency 증가
ingestion time:  Flink에 들어온 시점, 둘 사이의 절충

4.2. Stateful Stream Processing

• API는 functional하고 side-effect-free해 보이지만 실제로는 stateful computation을 지원
• state 예: ML model, graph analysis, user session, window aggregation
  • counter, sum처럼 단순할 수도 있고, classification tree, large sparse matrix처럼 복잡할 수도 있음
• window도 record를 bucket에 넣고 계속 업데이트하니까 stateful operator
• state를 API에서 명시적으로 관리하고, StateBackend로 어디에 저장하고 어떻게 checkpoint할지 설정
• 등록된 state는 checkpoint로 durable하게 저장되고 exactly-once update가 보장됨

4.3. Stream Windows

• unbounded stream을 유한한 logical view로 나눈 것이 window
• stateful operator로 구현되고 세 요소로 구성

Assigner → record가 어느 window에 속하는지 (sliding이면 여러 개일 수도)
Trigger  → window 연산을 언제 실행할지
Evictor  → window 안에서 어떤 record를 버릴지

• 조합해서 time, count, session, delta, landmark window 등을 표현
• e.g.

stream
    .window(SlidingTimeWindows.of(Time.of(6, SECONDS), Time.of(2, SECONDS)))
    .trigger(EventTimeTrigger.create())

• size 6초, slide 2초. watermark가 window 끝을 지나면 계산

stream
    .window(GlobalWindow.create())
    .trigger(Count.of(1000))
    .evict(Count.of(100))

• global window 하나에서 1000개마다 계산, 최근 100개만 유지
• stream을 먼저 key로 partition하면 window 연산이 worker마다 local하게 처리되므로 worker 간 coordination이 필요 없음

4.4. Asynchronous Stream Iterations

• incremental ML, reinforcement learning, graph approximation 등에서 loop가 필요
• streaming의 feedback loop는 대부분 coordination이 필요 없어서 asynchronous iteration을 지원
• feedback stream도 operator state의 일부로 취급해 global snapshot에 포함

5. BATCH ANALYTICS ON TOP OF DATAFLOWS

• bounded dataset은 unbounded stream의 특수한 경우라, 이론상 input 전체를 window 하나에 넣으면 batch가 됨
• 그래도 batch만의 장점이 있음: 더 간단한 API, 추가 최적화, fault tolerance bookkeeping 감소, staged scheduling
• 그래서 batch는
  • stream과 같은 runtime 사용
  • 필요하면 blocking stream으로 stage 분리
  • periodic checkpoint 대신 lost partition을 materialized intermediate stream에서 replay
  • sort 같은 blocking operator는 input 전체를 consume한 뒤 실행하고, memory가 부족하면 disk로 spill
  • DataSet API와 query optimizer

5.1. Query Optimization

• parallel DB 기법 기반: plan equivalence, cost model, interesting-property propagation
• 하지만 UDF가 많은 임의의 DAG라서 그대로 적용하기 어려움. UDF 내부 semantic이 안 보이니까 cardinality와 cost를 추정하기 어려움
• execution strategy: repartition, broadcast, sort-based grouping, sort join, hash join
• 여러 physical plan 중 network I/O, disk I/O, CPU 비용으로 선택. 추정이 어려우면 programmer hint를 받음

5.2. Memory Management

• JVM heap에 객체를 잔뜩 만드는 대신 memory segment에 serialized form으로 저장
• sort, join도 가능한 한 binary representation 위에서 직접 수행
• 그래서 (de)serialization overhead와 GC overhead가 줄고, disk spill이 가능하고, cache efficiency가 높아지고, memory pressure에서도 안정적

5.3. Batch Iterations

• graph analytics, gradient descent에는 BSP, SSP 같은 parallel iteration model을 씀
• iteration-control event로 superstep의 시작과 종료를 표시
• delta iteration: sparse computational dependency를 활용

6. RELATED WORK

• batch: MapReduce, Dryad, Tez, Drill, Impala, Spark
  • Spark는 DAG execution engine, SQL optimizer, driver-based iteration, micro-batch streaming의 조합
• streaming: Storm, Samza, MillWheel …
• 논문이 말하는 당시 Flink의 차별점

1. batch와 stream 모두 pipelined streaming execution
2. lightweight checkpointing 기반 exactly-once state consistency
3. native iterative processing
4. out-of-order 처리를 포함한 정교한 window semantic

• event time과 out-of-order 처리, exactly-once managed state, high throughput과 low latency를 동시에 준다는 게 주장

7. CONCLUSION

• Flink는 stream analytics와 batch analytics를 모두 하는 universal dataflow engine
• operator state와 logical intermediate result를 first-class citizen으로 취급
• batch API와 streaming API가 같은 runtime을 서로 다른 parameter와 execution strategy로 공유
• batch는 이론상 streaming의 특수한 경우지만, 실제 실행에선 optimizer, blocking operator, disk spill 같은 최적화를 얹음