플링크 공식문서 정리-Stateful Stream Processing

State 란?

데이터 스트림 처리에서 State 란 스트림으로 들어오는 데이터를 각각 독립적으로 처리하는 것이 아니라, 여러 데이터에 걸쳐서 정보를 기억하고 그에 따라 처리하는 것을 의미한다. 아래는 state와 관련된 예시이다

• 애플리케이션이 특정 이벤트 패턴을 검색할 때, state 는 지금까지 발생한 이벤트를 순서대로 저장
• 분/시간/일 단위로 이벤트를 집계할 때 state 는 보류 중인 집계를 보유
• 데이터 스트림에 대해 머신 러닝 모델을 학습시킬 때, state는 모델 매개변수의 현재 버전 보유

플링크는 데이터 스트림에서 state 를state를 가지고 있고, 장애상황에서도 이 state를 정상적으로 복구하기 위해 checkpoints와 savepoints라는 개념을 가지고 있다. 또한 state 정보는 플링크로 하여금 플링크 애플리케이션을 rescaling 하는 것을 가능하게 한다

State Persistence

플링크는 stream replay 와 checkpointing을 통해 내결함성을 구현한다

checkpoint는 input sources의 특정 포인트와 각 operators의 상태의 조합의 snapshot을 만드는 행위이다. 예를 들어 2개의 input stream으로부터 A operator, B operator를 거친 후 target source에 저장하는 플로우를 상상해 보자. 이런 플로우에서 checkpoint를 남긴다면 아래의 정보들이 snapshot에 기록될 것이다

• checkpoint를 남기는 시점의 1번 input stream의 데이터 위치
• checkpoint를 남기는 시점의 2번 input stream의 데이터 위치
• checkpoint를 남기는 시점의 A operator의 state
• checkpoint를 남기는 시점의 B operator의 state

checkpoint를 주기적으로 실행시켜 snapshot을 저장해 두면 플링크가 여러 장애상으로 인해 실패해도 최신 checkpoint 시점으로 복구한 다음, 그 부분부터 처리를 다시 시작할 수 있다

Barriers

플링크의 분산 snapshotting 에서 핵심 요소는 barrier이다. 이 barrier는 데이터 스트림에 주입된 후 다른 데이터들과 마차간지로 데이터 스트림에 흘러간다

그렇다면 이 barrier 을 통해 어떻게 snapshot을 만든 것일까? 과정을 간략히 요약하면 아래와 같다

1. 주기적으로 checkpoint 시점이 되면 input stream 으로부터 데이터 대신 barrier를 스트림에 넣어준다. barrier 에는 snapshot id 가 있다. 이 id를 n이라 하자. 그리고 barrier 가 주입되는 순간 input stream 이 최근에 방출했던 데이터의 position을 checkpoint coordinator (플링크의 JobManager) 한테 넘겨준다. 만약 barrier 주입 직전에 124번 position의 데이터가 스트림으로 방출 됐었다면 checkpoin coordinator에 124를 넘겨준다
2. 방출된 barrier 가 스트림을 따라 흐르다가 intermediate operator 를 만나면 opertaor의 현재 state를 snapshot으로 저장한다
3. snapshot을 저장한 후 barrier는 다시 하위 스트림으로 방출된다. 그리고 intermediate operator를 만날 때마다 [2]의 과정을 수행한다
4. barrier 가 sink operator 를 만나면 sink operator는 checkpoint coordinator에게 snapshot n 이 ack 됐다는 신호를 보낸다
5. snapshot n 이 완료됐으며, 플링크가 장애상황으로 재시작하면 snapshot n 부터 재시작하게 된다

주의사항

1. barrier 는 앞서 방출된 데이터를 앞서갈 수 없다. 즉 순서를 엄격하게 지켜야 한다. 그래야만 알맞은 state를 snapshot에 저장할 수 있을 것이다
2. 1개의 intermediate operator 가 2개의 input source 에서 barrier를 받는 상황에서 state를 snapshot에 저장하기 위해서는 2개의 barrier를 모두 기다려야 한다. 먼저 도착한 barrier 및 그 이후에 도착한 데이터들은 나머지 barrier 가 도착할 때까지 처리될 수 없다. 그래야만 알맞은 state를 snapshot에 저장할 수 있을 것이다

 

참고

https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/concepts/stateful-stream-processing/