[Flink] Flink Architecture

최근 진행하고 있는 프로젝트에 pyflink를 도입하기 위하여 기술 연구를 진행하고 있습니다.

flink 공식문서를 보고 학습 및 번역한 내용을 정리하였습니다.

 

Flink Architecture

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

Anatomy of a Flink Cluster

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

Client

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

JobManager

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

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

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

1. 다음 Task 혹은 Task set를 스케줄링 할 시기를 결정함
2. 완료된 Taks 또는 실행 실패에 대해 대응
3. Checkpoint를 조정
4. 실패에 대한 복구를 조정

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

• 고가용성을 고려시에 Kubernetes, Zookeeper를 활용함

ResourceManager

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

Dispatcher

• Flink WebUI를 실행하여 task execution에 대한 정보를 제공함
• Flink Application에 실행을 submit하기 위하여 REST interface를 제공
• submit된 각 job에 대해 새로운 JobMaster를 시작함

JobMaster

• 단일 JobGraph의 실행을 관리하는 역할
• 여러 작업은 각각의 고유한 JobMaster를 가지고, Flink Cluster에서 동시에 실행될 수 있음

TaskManagers

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

• taskmanager는 하나 이상 존재해야함
• 리소스 스케줄링의 최소 단위는 Task slot
• task slot의 수는 동시 처리 작업의 수를 나타내며, 한 task slot에서 여러 operator가 실행될 수 있음

Tasks and Operator Chains

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

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

Task Slots and Resources

• 각 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이 작업의 전체 파이프라인을 고정할 수 있고, 두가지 이점이 있음

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

Flink Application Execution

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 Cluster

Cluster Lifecycle

• 하나의 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을 따라감

Resource Isolation

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

Flink Session Cluster

Cluster Lifecycle

• Client는 여러 job을 기존의 cluster에 연결함
• 모든 job이 완료된 후에도 session이 수동으로 중지될 때까지 cluster 및 jobmanager는 계속 실행 됨
• Flink Session Cluster의 lifecycle은 flink job에 국한되지 않음

Resource Isolation

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

Other consierations

• 기존 클러스터가 존재하면, 리소스를 신청하고, TaskManager를 시작하는데 시간이 절약됨