03-2. 아파치 플링크 아키텍처

상태 관리

• 보통 태스크의 상태는 태스크가 유지하고 있는 데이터와 함수의 결과를 게산할 때 사용하는 모든 데이터 포함
  • 그림 3-10. 상태가 있는 스트림 처리 태스크
  • 상태 일관성과 관련된 모든 문제, 실패 처리와 저장소의 효과적인 접근 등은 플링크가 책임지므로 개발자는 애플리케이션 로직에만 집중 가능
  • 상태의 종류 : 연산자 상태(operator state), 키 상태(keyed state)
    • 스코프(scope)에 따라 접근 제한됨

 

 

연산자 상태

• 스코프 : 연산자 태스크 하나
  • 한 태스크가 처리하는 모든 레코드가 동일 상태에 접근 가능
  • 그림 3-11. 연산자 상태를 가진 태스크
  • 플링크가 제공하는 세 종류의 기본 연산자 상태
    • 리스트 상태(List state)
      • 리스트의 요소들로 상태 표현
    • 유니온 리스트 상태(Union List state)
      • 리스트의 상태를 나타내는 것은 같으나 장애를 복구하거나 애플리케이션을 세이브포인트에서 시작할 때 리스트 상태와 차이 있음
    • 브로드캐스트 상태(Broadcast state)
      • 연산자의 모든 태스크 상태가 동일한, 특수한 경우에 사용하는 연산자 상태
        • 체크포인팅을 수행하거나 연산자를 재확장(rescaling)할 때 사용

 

 

키 상태(keyed state)

• 레코드의 각 키 값별로 유지하고 접근할 수 있는 상태
• 플링크는 각 키 값별로 하나의 상태 인스턴스 유지. 이 키 상태를 관리하는 연산자 태스크로 동일 키의 모든 레코드 전송
  • 태스크가 어떤 레코드를 처리할 때 현재 처리중인 레코드의 키와 일치하는 키 상태만 접근 가능 -> 동일 키를 가진 모든 레코드는 같은 상태에 접근
• 그림 3-12. 키 상태를 사용하는 태스크
• 키-값 맵(key-value map)처럼 키 값으로 연산자 병렬 태스크에 레코드를 분배한 것
• 플링크는 이 분산 키-값 맵에 키 별로 저장할 수 있는 여러 상태 값 종류 제공
  • 값 상태(Value state)
    • 임의 타입의 값을 키 별로 저장. 복잡한 데이터 구조도 값 상태로 저장 가능
  • 리스트 상태(List state)
    • 각 키별로 값 리스트 저장. 리스트의 요소는 어떤 타입이라도 가능
  • 맵 상태(Map state)
    • 키별로 키-값 맵을 저장. 맵의 키와 값은 어떤 타입이든 가능

 

 

상태 백엔드

• 연산자 태스크는 상태를 로컬에 유지해 고속의 상태 접근 보장
• 상태의 정확한 저장, 접근, 유지는 상태 백엔드(state backend)라 부르는 플러그인 가능한 컴포넌트가 결정
• 상태 백엔드는 로컬 상태 관리와 원격 저장소에 상태를 체크포인팅 하는 두 가지 책임 지고있음
• 플링크
  • JVM 힙(heap)의 메모리 데이터 구조인 객체(object)를 저장하는 것처럼 상태를 관리하는 메모리 상태 백엔드 제공
    • 고속의 상태 접근 제공, 메모리 크기에 제약
  • 상태 객체를 직렬화하고 이 데이터를 RocksDB에 저장
    • 상태 접근은 느리지만 상태 크기는 천천히 증가
• 태스크매니저 저장소는 프로세스가 언제라도 장애가 발생할 수 있으니 휘발성(volatile)이라 생각해야함
  • 상태 백엔드는 원격의 영구적인 저장소에 태스크의 상태를 체크포인팅함

 

 

상태가 있는 연산자의 수평 확장

• 상태가 있는 연산자의 병렬 값 변경은 상태를 재분할해(repartitioning) 늘어난 병렬 태스크에 다시 할당해야 하므로 어려움
• 플링크의 상태 수평 확장 지원의 네 가지 패턴
  • 키 상태를 사용하는 연산자
    • 수평 확장 시 병렬 태스크가 갖고 있는 키들을 더 적거나 많게 재분할
    • 키를 키 그룹(key group)으로 구조화. 하나의 키 그룹은 하나의 키 파티션. 플링크는 이 키 그룹 단위로 태스크에 여러 키 할당
    • 그림 3-13. 키 상태를 사용하는 연산자 수평 확장과 축소
  • 리스트 상태를 사용하는 연산자
    • 리스트 요소를 재분배해 확장
    • 연산자의 모든 병렬 태스크가 갖고 있는 리스트 요소를 수집해 새 태스크로 균등하게 재분배
    • 그림 3-14. 리스트 상태를 사용하는 연산자의 수평 확장과 축소
  • 브로드캐스트 상태를 사용하는 연산자
    • 상태를 새로운 태스크로 복사해 확장
    • 모든 태스크가 동일한 상태를 가졌다는 것을 보장하므로 이런 동작이 가능
    • 3-16. 브로드캐스트 상태를 사용하는 연산자의 수평 확장과 축소

 

체크포인트, 세이브포인트, 상태 복구

• 정확히 한번(exactly-once) 상태 일관성을 보장하려고 플링크가 사용하는 체크포인트와 복구 방식

 

 

일관성 체크포인트(consistent checkpoint)

• 모든 태스크가 정확히 동일한 시점에 각 태스크의 상태 복사
  1. 모든 입력 스트림 인입 금지
  2. 현재 처리 중인 데이터가 완료되기까지 대기
  3. 각 태스크의 상태를 복사해 원격의 영구저장소로 체크포인트 지정
  4. 모든 스트림의 인입 재시작
• 그림 3-17. 스트리밍 애플리케이션의 일관성 체크포인트

 

 

일관성 체크포인트에서 복구

• 플링크는 주기적으로 일관성 체크포인팅을 수행해 상태를 원격 저장소에 저장
• 장애 발생 시 가장 최신 체크포인트를 사용해 일관성 있게 애플리케이션 상태 복구. 레코드 처리 재시작.
• 그림 3-18. 체크포인트에서 애플리케이션 복구
• 애플리케이션 세 단계 복구
  1. 전체 애플리케이션 재시작
  2. 상태가 있는 모든 태스크 상태를 가장 최신의 체크포인트로 재설정
  3. 모든 태스크 처리 시작
• 모든 연산자가 체크포인팅 수행하고 모든 상태 복구한 후 모든 입력스트림의 위치가 체크포인팅 수행 당시 마지막으로 소비했던 위치로 재설정 된다면 애플리케이션 정확히 한번(exactly-once) 일관성 제공 가능
• 애플리케이션의 모든 입력 스트림이 재설정 가능한 데이터 소스에서 데이터 소비하면 정확히 한 번 상태 일관성 운영 가능(ex. 아파치 카프카)
• 상태 복구 후 애플리케이션은 체크포인팅 수행과 장애로 처리 중단된 시간 동안 쌓여있던 모든 데이터 재처리
  • 모든 연산자 상태가 이 데이터들을 처리하지 않은 시점으로 재설정되어서 정확히 한번 상태 일관성 보장

 

플링크의 체크포인트 알고리즘

• 스트림 애플리케이션 전체에서 동시에 체크포인팅을 수행하는 단순 접근 방식은 짧은 지연 요구하는 애플리케이션에선 사실상 사용 불가능
  • 스탑더월드(stop-the-world) 일으킴 (gc 실행 위해 jvm이 애플리케이션 실행 멈춤)
• 플링크는 챈디-램포트(Chandy0Lamport) 알고리즘을 기반으로 분산 스냅샷(distributed snapshot) 체크포인팅 구현
  • 애플리케이션 전체를 정지하지 않고 체크포인트와 데이터 처리 간의 결합 분리
  • 일부 태스크는 상태 저장하고 일부는 데이터 계속 처리
  • 배리어(barrier) 레코드 사용
    • 소스 연산자가 레코드 스트림에 체크포인트 배리어 주입
    • 체크포인트 배리어 이전에 처리한 레코드가 만든 상태 변경은 배리어의 현재 체크포인트에 포함. 이후는 다음 체크포인트에 포함
• 그림 3-19. 상태가 있는 소스 두 개, 상태가 있는 태스크 두 개, 상태가 없는 싱크 두 개를 가진 스트리밍 애플리케이션
• 그림 3-20. 잡매니저는 모든 소스로 메시지를 보내 체크포인트 초기화
• 그림 3-21. 소스는 자신의 상태를 체크포인팅하고 체크포인트 배리어 내보냄
  
  • 체크포인트 메시지 수신 시 레코드 내보내지 않음
  • 소스 태스크는 로컬 상태를 상태 백엔드에 체크포인팅하고 체크포인트를 배리어 밖으로 향하는 모든 스트림 파티션으로 브로드캐스트
  • 상태 체크포인팅 완료시 태스크에 알리고, 태스크는 잡매니저에게 알림
  • 모든 배리어 전송 후 소스는 자신의 연산 계속 수행
  • 입력스트림은 체크포인팅이 수행됐던 시점의 위치에서 다시 시작
• 그림 3-22. 태스크는 각 입력 파티션에서 배리어를 수신하고자 기다림. 배리어가 이미 도착한 입력스트림의 레코드는 버퍼링됨. 배리어가 도착하지 않은 입력 스트림의 레코드는 계속 처리됨
  
  • 배리어 정렬(barrier alignment) : 모든 배리어가 도착하기를 기다리는 것
• 그림 3-23. 모든 배리어를 받자마자 자신의 상태를 체크포인티하고 체크포인트 배리어를 내보내는 태스크
• 그림 3-24. 체크포인트 배리어를 내보낸 후 원래의 처리를 계속 수행하는 태스크
• 그림 3-25. 싱크가 체크포인트 배리어의 수신 여부를 잡매니저에 알리고 모든 태스크가 자신의 상태 체크포인팅이 성곶적으로 끝났음을 잡매니저에 알리면 체크포인팅 완료

 

 

체크포인트가 성능에 미치는 영향

• 플링크 설계상 체크포인팅 수행은 상태 백엔드의 책임. 상태 백엔드 구현에 따라 태스크의 상태 복제하는 방식 다름
  • 파일 시스템(FileSystem) 상태 백엔드와 RocksDB 상태 백엔드는 비동기 체크포인팅 지원
    • 체크포인팅 시작되면 상태 백엔드는 상태를 로컬에 복사. 복사 완료시 태스크 하던 일 계속 진행
  • 백그라운드 스레드는 로컬 스냅샷을 원격 저장소에 비동기로 복사. 체크포인팅 완료되면 태스크에 알림
    • 비동기 체크포인팅은 태스크의 데이터 처리 재시작까지 걸리는 시간 줄여줌
    • RocksDB 상태 백엔드는 증분 체크포인트(Incremental checkpoint) 기능도 있어서 전송할 데이터양 많이 줄일 수 있음
• 배리어 정렬(Barrier Alignment) 단계 변형
  • 매우 짧은 지연과 최소 한 번(at-least-once) 상태 보장으로 충분한 애플리케이션은 배리어 정렬 시 배리어 도착 후 들어오는 레코드를 버퍼에 쌓는 대신 계속 처리하도록 플링크에 설정

 

 

세이브포인트(savepoint)

• 체크포인트에 몇 가지 메타데이터를 추가한 것
• 사용자가 명시적으로 세이브포인트 생성해야 함. 자동 삭제x

 

 

세이브포인트 사용

• 애플리케이션을 세이브포인트에서 시작 가능(애플리케이션과 호환성 가진 세이브포인트- 애플리케이션이 세이브포인트 상태 읽기 가능)
  • 세이브포인트와 호환되는 다른 애플리케이션 시작 가능
  • 애플리케이션 로직 버그 수정, 상태 복구 후 입력 스트림의 이벤트 재처리 가능
• 동일 애플리케이션을 다른 병렬 값으로 시작하거나 규모 확장, 축소 가능
• 다른 클러스터에서 동일 애플리케이션 실행 가능
• 세이브포인트 이용해 애플리케이션 정지시키고 나중에 재시작 가능
• 애플리케이션의 상태를 이력으로 남기거나 아카이빙 가능

 

 

세이브포인트에서 애플리케이션 시작

• 실행중인 애플리케이션의 세이브포인팅 수행 후 애플리케이션 시작 시 상태 복구에 세이브포인트 사용
• 그림 3-26. 애플리케이션에서 세이브포인팅 수행하고 이 세이브포인트에서 애플리케이션 복구
  
  • 세이브포인트 실행되면 모든 태스크의 상태를 영구 저장소로 복사
  • 세이브포인트의 상태 데이터는 연산자 식별자(ID)와 상태이름 가지고 있음
    • 세이브포인트와 관련된 연산자가 있는 태스크로 세이브포인트 데이터 재분배
  • 수정한 애플리케이션을 세이브포인트에서 시작한다면 애플리케이션이 세이브포인트와 일치하는 식별자와 상태 이름을 가진 연산자를 포함할 때만 세이브포인트의 상태 복구 가능
  • 플링크는 연산자에 고유 식별자 설정되어 있지 않으면 자동으로 할당
    • 연산자 식별자는 앞쪽의 연산자 식별해 생성됨 - 앞쪽의 연산자 식별자 변경시 현재 연산자 식별자도 변경됨
    • 상태를 유실해도 관계없는 애플리케이션일 때만 기본 연산자 식별 사용하도록 제한해야 함