Haonly's Blog

에러 메시지

Caused by: java.lang.AssertionError: assertion failed: Concurrent update to the commit log. Multiple streaming jobs detected for

원인

로그에 동시 업데이트 하기 때문에 발생

스파크 스트리밍에서 동일한 체크포인트 위치에 두 개 이상의 다른 스파크 작업이 업데이트 하려 할 때 발생한다.

이 때 스파크 설정의 checkpointLocation을 다른 위치로 사용하면 해결할 수 있다.

두 개의 다른 스파크 스트리밍 작업에서 checkpoint 위치를 다르게 한다.

      .option("checkpointLocation", checkpointPath1)

      .option("checkpointLocation", checkpointPath2)

카카오 제네시스 - 카프카 기반 스트리밍 데이터 플랫폼

2021년 Kakao 에서 봤던 Cory 님(광고추천팀 - 데이터 플랫폼 개발)이 작성한 글이고 내가 요즘 공부하는 카프카, 업무와 직관된 데이터 플랫폼이라는 제목 워딩에 끌려 클릭했다.

개인적으로 사용자에게 직접 서비스될 수 있는 분야가 업무적으로 더 선호된다. 하지만 이미 플랫폼에 집중된 내 업무로는 직접 서비스할 기회는... 적다. 거의 없을 수도..

그런데 광고 추천이라니. 내가 직접 서비스하진 않더라도 결국에는 개인화된 광고를 서빙하는 작업을 하는 플랫폼일 것이다! 완전 끌린다.

(나는 누가 쓰는지도 모르고.. 그냥 들어오는 데이터 ETL 하는 느낌인데 말이다.. ㅠㅠ)

아무튼 그래서 이번 기술 블로그 엄청 재밌게 읽었다! 공부 의욕 뿜뿜!

들어가면 나오는 내용이지만 나만을 위한 정리

• 기존 카프카 데이터 파이프라인 아키텍처의 관리에서의 어려움과 리소스 낭비로 새로운 카프카 커넥트 기반 데이터 플랫폼을 구성
• 고려한 점: 오너십 / 모니터링 / 배포 / 데이터 리니지(화면)
• 카프카 커넥트 사용
  • ETL 역할을 수행하는 것을 커넥터라고 하며, 싱크 커넥터는 consumer, 소스 커넥터는 producer 역할을 한다.
  • 카프카 커넥트는 분산 커넥트와 단일 커넥트로 나뉜다.

• 카프카 커넥트를 API 동작이 아닌, 지속적 운영을 위해 vue.js로 어드민 페이지를 만들어서 모든 파이프라인 관련 동작을 제네시스 웹을 통해 수행 가능하록 개발.
• 카프카 커넥트를 운영하며 고려해야 할 점
  • 반드시 웹 화면이 필요
    • REST API를 통해 파이프라인을 생성, 수정, 삭제할 수 있지만 언제까지나 API 툴로 운영할 수가 없음
    • 오픈소스로 나와 있는 카프카 커넥트 웹을 사용해도 됨
  • 커스텀 커넥터 개발 → 보안 관련 이슈
  • 커넥터 클러스터 구분 운영
    • 커넥터의 특성(메몰리 많이 사용/CPU 많이 사용)에 따라 커넥트 클러스터를 분리하여 운영하는 것을 고려

도커로 말아서 올렸더라. 도커도 또 공부하려면 한참인데 정말...!!

이거 다 개발하려면 진짜 많은 시간과 노력이 들었겠다 싶다.

나도 하고 싶다. 배울게 너무 많다...!

한 줄 느낀 점:
플랫폼 운영하려면 알아야 할게 많다. 좋게 생각하면 배우는 거 좋아하는 나한테 딱!

카프카 아키텍처 생각해 보다가 갑자기 궁금한 점이 생겼다.

카프카는 같은 파티션 안에서 순서를 보장해주며 여러 파티션을 분산 처리할 수 있는 장점을 가지고 있기 때문에 사용한다.

근데 "같은 파티션 안에서 순서를 보장해준다면 같은 토픽의 다른 파티션 간의 순서는?" 이라는 의문이 들었다.

마침 우아한형제들에서 진행하는 우아한테크세미나에서 이 의문에 대해 의외로 간단한 답변으로 답을 듣게 되었다.

다른 파티션 간의 순서는 보장 안한다.

물론 아주 복잡하게 순서가 뒤바뀌어 들어와 처리 순서가 많이 뒤바뀌진 않기도 하겠지만, 다른 파티션 간의 순서까지 보장하려면 FIFO 쓰지 왜 카프카 쓰겠느냐는 것이다. 

음식 주문의 예시로 생각해보면 1분에 1000개 주문이 들어온다고 할 때, 그 안에서의 순서는 크게 서비스에 영향을 미치지 않는다는 것 같다.

파티션이 4개라면 그 안에서 250개의 주문들은 각 파티션 안에서 순서가 지켜질 것이다.

카프카를 나름 잘 안다고 생각하고 있는데, 공부하면 할수록 공부할 것들이 많다...!

카프카 스트림즈라는 카프카 내의 새로운 개념까지 최근에 알아버려 공부할 게 또 생겼다.

재미있기는 한데 나는 왜이리 모르는 게 많은 건지ㅠㅠ ㅋㅋㅋ 시간을 쪼개어 공부를 열심히 해야겠다! 

Kafka 정의

• Distributed commit log
• 데이터를 지속해서 저장하고 읽을 수 있으며 확장에 따른 성능 저하를 방지하여 데이터가 분산 처리될 수 있다.

메시지 발행과 구독

• publish/subscribe 관계로 데이터 발행자가 발행/구독 시스템에 전송하면 구독자가 특정 부류의 메시지를 구독할 수 있도록 함
• 메시지를 저장하고 중계하는 역할은 브로커가 수행
• 메시지 발행 시스템에서즌 데이터 발행자가 직접 구독자에게 보내지 않는다!

초기의 발행/구독 시스템

• 발행자와 구독자가 직접연결된 시스템에서 발전하여
• 중간 브로커(서버)를 둔 아키텍쳐가 만들어졌다.
• 모든 애플리케이션의 메트릭을 하나의 애플리케이션이 수신하고, 하나의 서버로 제공하여 해당 메트릭이 필요한 어떤 시스템에서도 쉽게 조회 가능

카프카 사용 이유

• 중앙화된 전송 영역이 없어 end to end 연결이 복잡해진다
• 데이터 파이프라인 관리가 어려워지고
• 연결 시스템마다 다른 방식으로 구현 수 있다

메시지와 배치

• 데이터의 기본단위: 메시지
• 바이트 배열의 데이터로 특정 형식이나 의미를 갖지 않음
• 카프카 메시지 데이터는 topic으로 분류된 partition에 수록되는데 이때 데이터를 수록할 파티션을 결정하기 위해 일관된 해시 값으로 키를 생성한다.
• 카프카는 메시지가 생길때마다 보내는 게 아닌 효율성을 위해 모아 전송하는 형태의 배치로 파티션에 전송할 수 있다.
  • 배치의 크기가 증가하면 → 단위 시간당 처리되는 메시지 양이 증가하지만 메시지의 전송 시간도 증가함

스키마

• 메시지의 구조를 나타내는 스키마를 사용
• Json / XML 등을 사용
• 스키마 버전 간의 호환성이 떨어져 아파치 Avro를 많이 사용(Apache에서 만듦)

토픽과 파티션

• 카프카의 메시지는 토픽(topic)으로 분류하며 토픽은 DB의 테이블이나 파일 시스템의 폴더와 유사
• 하나의 토픽은 여러개의 파티션(partition)으로 구성됨
• 메시지는 파티션에 추가되는 형태로만 수록
• 맨 앞부터 끝까지 순서대로 읽힌다
• 메시지의 처리 순서는 토픽이 아닌 파티션 별로 유지 관리
• 각 타피션은 서로 다른 서버에 분산되어 단일 서버로 처리할 때보다 훨씬 성능이 우수하다.

프로듀서와 컨슈머

• 데이터를 쓰는 프로듀서(producer) - 데이터를 읽는 컨슈머(consumer)
• 오프라인으로 대량의 데이터를 처리하도록 설계된 프레임워크인 하둡의 방법과 대비된다.
• 프로듀서: 새로운 메시지를 생성하며 메시지는 특정 토픽으로 생성되고 기본적으로 메시지가 어떤 파티션에 수록되는지는 관여하지 않음
  • 프로듀서가 특정 파티션에 메시지를 직접 쓰는 경우가 있음 - 파티셔너 사용(키의 해시 값을 생성하고 그것을 특정 파티션에 대응시켜 지정된 키를 갖는 메시지가 항상 같은 파티션에 수록되게 함)
• 컨슈머: 메시지를 읽는 주체
  • 하나 이상의 토픽을 구독하여 메시지가 생성된 순서대로 읽는다.
  • 메시지의 오프셋을 유지하여 메시지의 읽는 위치를 알 수 있다.
  • 오프셋은 지속적으로 증가하는 정수값이며 메시지가 생성될 때 카프카가 추가해준다.
  • 파티션에 수록된 각 메시지는 고유한 오프셋을 갖는다.
  • 주키퍼나 카프카에서는 각 파티션에서 마지막에 읽은 메시지의 오프셋을 저장하고 있으므로 컨슈머가 메시지 읽기를 중단했다 다시 시작해도 언제든 다음부터 읽을 수 있다.

• 컨슈머는 컨슈머 그룹의 멤버로 동작
• 한 토픽을 소비하기 위해 같은 그룹의 여러 컨슈머가 함께 동작
• 한 토픽의 각 파티션은 하나의 컨슈머만 소비할 수 있다.
• 한 컨슈머가 자신의 파티션 메시지를 읽는데 실패해도 같은 그룹의 다른 컨슈머가 파티션 소유권을 재조명받고 실패한 컨슈머의 파티션 메시지를 대신 읽을 수 있다.

브로커와 클러스터

• 브로커: 하나의 카프카 서버
• 프로듀서로부터 메시지를 수신하고 오프셋을 저장한 후 해당 메시지를 디스크에 저장
• 클러스터의 일부로 동작하도록 설계되었다
• 컨트롤러의 기능을 수행
• 각 타피션은 클러스터의 한 브로커가 소유하며, 그 브로커를 파티션 리더라고 한다.
• 일정 기간 메시지를 보존하는 기능도 있다
• 각 파티션을 사용하는 모든 컨슈머와 프로듀서는 파티션 리더에 연결해야 한다.
• 다중 클러스터
  • 데이터 타입에 따라 구분 및 처리
  • 요구사항에 따라 분리 처리
  • 다중 데이터 센터 처리
  • 다중 클러스터를 지원하기 위해 미러 메이커를 사용한다.

카프카를 사용하는 이유

• 다중 프로듀서, 컨슈머
  • 다중 프로듀서, 컨슈머가 상호 간섭 없이 어떤 메시지 스트림도 읽을 수 있다.
• 디스크 기반 보존
  • 메시지를 보존할 수 있어 컨슈머를 항상 실시간 실행시키지 않아도 된다.
  • 처리가 느리거나 접속이 폴주해서 메시지 읽는데 실패해도 데이터가 유실될 위험이 적다
• 확장성
  • 브로커 1대부터 시작하여 규모에 따라 브로커를 수백대로 증가시키기고 대규모 클러스터로 묶어 사용할 수 있다.
  • 동시에 여러 브로커에 장애가 생겨도 복제 팩터를 더 큰 값으로 했다면 대응할 수 있다.
    
    

이용 사례

• 활동 추적
• 메시지 전송(메일, 푸시 알림)
• 메트릭 로깅
• 커밋 로그
• 스트림 프로세싱

Kafka consumer group 모니터링 중 "Preparing rebalance" 상태가 떠서 정확히 알아볼 겸 consumer state 에 대한 정리...

(기초는 차차 정리하고 일단 당장 궁금한 것부터!)

Rebalancing 정의

• 카프카 컨슈머 그룹을 구성하여 데이터를 처리할 때 운영상황에서 마주치는 다양한 경우의 수에 대해
• 그룹에 참여하는 컨슈머 클라이언트 구성에 변화가 생길 경우 이 변화를 반영하기 위한 과정
• 컨슈머 클라이언트를 사용하여 필요한 논리적인 그룹을 형성하고, 그 그룹 멤버들끼리 리소스(파티션)을 적절히 분배하는 프로세스

Consumer Group

• 특정 토픽에 발행된 메시지를 consumer group이라는 논리적 멤버십을 제공함으로서 각 그룹에서 목적에 맞게 읽고 처리할 수 있도록 함
• consumer group과 그 멤버 인스턴스, 토픽의 파티션과의 관계는 복잡함
• 컨슈머 인스턴스 제약
  • 카프카에서 1개 파티션은 consumer group 내의 최대 1개 인스턴스까지만 접근 가능
  • 즉, 1개 consumer group 인스턴스 수가 토픽이 가진 파티션 갯수보다 많을 수 없음
  • 다시 적자면, 파티션 수가 consumer group내에서 실제 컨슈밍을 진행하고 있는, 최대 active consumer 갯수를 제한함
  • (consumer group과 파티션 사이의 관계가 있다는 것이며 이런 맥락에 기반하여 분산 처리 및 로드 밸런싱이 이루어짐)

Kafka Rebalance Protocol

• 토픽을 컨슈밍할 논리적인 그룹을 형성하고 파티션을 재분배하는 과정
• Group management를 사용하여 카프카 클라이언트들을 논리적인 그룹에 참여시킴으로서 cooperating 하며
• Cooperating 과정에서 그룹 코디네이터(Group Coordinator)는 GroupCoordinator 인스턴스를 백그라운드 프로세스로 실행하며 consumer group을 관리하는 역할을 가진 카프카 브로커임
• Kafka Rebalancing Protocol에서 코디네이터의 역할 대신 클라이언트 단의 역할이 크게 작용하게 되며, 이에 따른 장점이 있음
  • 클라이언트 자체로 로드밸런싱 알고리즘을 수행: autonomy → 클라이언트가 로드벨런싱 자체의 디테일 작업에 집중하며 브로커의 코드가 단순화됨

카프카 토픽 파티션 갯수 설정

토픽을 새로 생성했는데 파티션을 몇개로 설정할지 고민에 빠졌다.
파티션 갯수를 늘리면 로드가 줄어들기는 하지만 늘리는 만큼 복제 지연 이슈가 있을 수 있으니 적당히 나눠야 한다.

바람직한 갯수는 아래와 같다.
#partitions = desired throughput / partition speed

예를 들어 하루에 1TB 의 throughput을 원한다고 하고 하나의 파티션 속도가 10MB/s 라면
1TB/24h = 11.5MB/s 이므로
대략 1개~2개 정도면 원하는 throughput을 만족할 수 있는 속도가 나온다.