Flink Architecture

Flink 분산 시스템이며 스트리밍 애플리케이션을 실행하려면 컴퓨터 리소스의 효과적인 할당 / 관리가 필요함
Hadoop YARN 및 Kubernetes 와 같은 클러스터 리소스 관리자와 통합됨
Standalone cluster 또는 라이브러리로 실행되도록 설정할 수 있음
- Standalone cluster는 머신에서 직접, container로 혹은 Yarn과 같은 리소스 프레임워크로 관리됨
TaskManager는 Jobmanager에 연결하여 자신이 사용가능한 것으로 알리고 JobManager에서 작업을 할당받음

Anatomy of a Flink Cluster

Flink은 런타임은 JobManager와 하나 이상의 TaskManager라는 두가지 유형의 프로세스로 구성됨

런타임 및 프로그램 execution의 일부는 아니지만 dataflow를 준비하고 Jobmanager로 보내는데 사용됨
- 실행을 트리거하는 Java/Scala 프로그램의 일부로 실행되거나, ./bin/flin run … 과 같은 명령문으로 실행됨
- pyflink로 작성하여, job을 던질 때 execution에서 실패 -> dataflow(jobgraph)를 그리지 못한 Error가 종종 나왔음
연결을 끊거나(detached mode) 연결을 유지(attached mode)하여 진행 report를 받을 수 있음

Flink 애플리케이션의 Coordinating of distributed execution(분산 실행을 조정하는 것?)에 책임이 있음.

→ 여러 TaskManager에 적절하게 job을 분배하는 것으로 해석하였음

→ Trino coordinator, Spark Master node와 유사한 구조로 해석하였음

하나 이상의 JobManager를 가지며, HA(고가용성)에서는 여러 JobManager를 가질 수 있음

jobmanager의 일부 component로 Flink 클러스터에서 리소스 할당 / 해제, 프로비저닝을 담당 → Task Slot을 관리
- Task Slot : Flink 클러스터의 리소스 스케줄링 단위
Flink는 Yarn, Kubernetes 또는 standalone 배포와 같은 다양한 환경이나 Resource Provider를 위해 여러 ResourceManager를 구현함
- 여러 Resource Provider(리소스 공급자)에 대응이 가능하다는 의미로 해석함
standalone setup에 경우 resource manager는 taskmanager의 사용가능한 slot에만 배포할 수 있으며, 새로운 taskmanager를 시작할 수는 없다.

worker라고도 하며 dataflow에서 task를 실행하고 data streams을 교환하고 버퍼링을 한다.

분산 실행을 위해 flink chain operator는 subtasks를 함께 수행함
각 task는 하나의 스레드에 의해 실행됨
operator의 task를 함께 chaining 하는 것은 유용한 optimization이다.
- 스레드간 handover 와 buffering을 줄이고, 전체 처리량을 늘리는 동시에 지연 시간을 줄임
chaining behavior을 구성할 수 있음

아래의 sample data flow는 5개의 subtask로 실행되며. 5개의 병렬 스레드로 실행됨.

각 worker(=taskmanager)는 jvm process이며, 하나 이상의 subtask는 개별 스레드에서 실행할 수 있음
taskmanager가 허용하는 task수를 제어하기 위하여 하나 이상의 taskslot이 있음
각 taskslot은 taskmanager에 리소스의 고정된 subset을 나타냄
- 3개의 task slot이 있는 taskmanager는 memory의 1/3을 각 slot에 할당
- 리소스를 slotting한다는 것은 subtasks는 memory에 대해서 다른 작업의 subtask와 경쟁하지 않고 일정량의 스케줄링된 memory가 있음을 의미함
- cpu isolation은 발생하지 않음
task slot의 수를 조정하여 사용자는 subtask가 서로 분리 되는 방식을 정의할 수 있음
taskmanager당 하나의 slot이 있다는 것은 각 task group이 별도의 JVM(ex: container)에서 실행된다는 것을 의미
taskmanager당 여러개의 slot이 있다는 것은 동일한 JVM을 공유하는 subtask가 더 많다는 것을 의미함
- 동일한 JVM 내의 task는 tcp connection과 heartbeat messages를 공유함 -> 병렬로 실행시 subtask간 hearbeat messages 전송하며 connection을 맺고, 상태확인을 하는것을 볼 수 있음.
data sets, data structure를 공유하여 task별 over-head를 줄일 수 있음

기본적으로, 다른 task의 subtask도 slot을 공유하도록 허용함

하나의 slot이 작업의 전체 파이프라인을 고정할 수 있고, 두가지 이점이 있음

flink cluster의 job에 사용되는 높은 parallelism만큼 taskslot이 필요함
slot 공유를 활용하게될 때 예제의 parallelism을 2개 → 6개로 늘린 것을 보아 slot 리소스를 완전하게 활용하는 동시에 무거운 subtask를 taskmanager간에 공평하게 분배함

Flink Application은 main() 메서드에서 하나 또는 여러개의 flink job을 생성하는 모든 사용자 프로그램

이러한 jobs는 로컬 JVM(Local Environment) 또는 여러 시스템이 있는 클러스터(Remote Environment)의 원격 설정에서 수행될 수 있음
각 프로그램에 대하여 Execution Environment는 job execution을 제어하고 외부와 상호작용하는 방법을 제공함

Flink Application은 Flink Session Cluster 혹은 Flink Application Cluster로 submit 될 수 있음

하나의 Flink Application에서 작업만 실행하는 전용 Flink 클러스터이며 main() 메서드는 Client가 아닌 클러스터에서 실행됨.
job submision은 한단계의 프로세스로 flink cluster를 시작하고, job을 기존의 cluster session에 submit할 필요가 없음
- 대신 Application logic과 종속성을 jar에 패키지하고 클러스터 진입점이 main() 메서드를 호출하여 JobGraph를 추출하는 역할을 함
Flink Application의 life cycle은 Flink Application Cluster life cycle을 따라감

Flink Application Cluster에서 ResourceManager와 Dispatcher는 single Flink Application으로 범위가 지정됨
- Flink Session Cluster보다 문제를 더 잘 분리할 수 있음

TaskManager slot은 ResourceManager에 의해 할당되고 job이 완료되면 해제됨
- 모든 job이 동일한 클러스터를 공유하기 때문에 submit 작업에서 네트워크 대역폭과 같은 클러스터 리소스에 대한 경쟁이 발생
중요한 점은 하나의 TaskManager가 충돌하면 이 TaskManager에서 실행 중인 Task가 있는 모든 job이 실패하고, 마찬가지로 JobManager에서 치명적인 오류가 발생하면 클러스터에서 실행 중인 모든 job에 영향을 주게됨