Haonly's Blog

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을 만족할 수 있는 속도가 나온다.

사내 스터디 세미나로 준비한 자료 정리

Go 언어를 활용한 마이크로서비스 개발

2장 좋은 API 디자인하기

api 규약 작성이 매우 중요하며 어려움

RESTful API

• REpresentational State Transfer(표현적 상태 전송)
  • 컴포넌트(서비스 단위) 간 상호작용의 확장성, 범용적인 인터페이스, 컴포넌트의 독립적인 배포를 강조하며 응답 지연시간 감소, 보안 강화, 레거시 시스템의 캡슐화를 위한 중간 컴포넌트 역시 강조한다.

URI, URN, URL

• RFC 3986

• 

• URI 형식 지정의 규칙
  • 슬래시(/)는 리소스 사이의 계층적 관계를 나타내는 데 사용
  • URI의 마지막에 슬래시가 포함되어서는 안된다.
  • 하이픈(-) 사용, _ 사용 x
  • 대소문자 구분하므로 소문자 사용 권장

URI 경로

표현설명비고

REST URI 디자인

• DELETE /cats/1234: 좋은 예
• GET /deleteCat/1234: 나쁜 예
• POST /cats/1234/delete: 나쁜 예
• HTTP 동사
  • GET / POST / PUT / DELETE / PATCH / HEAD / OPTIONS

응답코드

• 요청의 성공이나 실패를 클라이언트에게 알려주기 위한 HTTP 응답 코드
• 즉각적으로 요청의 상태를 알 수 있게 설계

나쁜 응답

POST /kittens RESPOSNE HTTP 200 OK { "status": 401, "statusMessage": "Bad Request" }

POST /kittens RESPOSNE HTTP 200 OK { "id": "123434jvjv4564", "name": "Fat Freddy's Cat" }

• 좋은 응답은 HTTP 상태 코드를 문자 그대로 사용하는 것

실패 응답의 좋은 예

POST /kittens RESPONSE HTTP 400 BAD REQUEST { "errorMessage": "Name should be between 1 and 256 characters...." }

성공한 응답의 좋은 예:

POST /kittens RESPONSE HTTP 201 CREATED { "id": "123434jvjv4564", "name": "Fat Freddy's Cat" }

• IBM HTTP 요청과 응답 예제

HTTP 상태 코드

• 구글에 더 자세하게 나와있지만 간략하게 정리
• 200 OK: 요청이 성공했음을 나타내는 일반적인 응답 코드
• 201 Created(생성): 요청이 성공하고 새 엔티티가 생성된 경우의 응답
• 204 No Content: 내용 없음. 클라이언트의 요청이 잘 처리되었지만 본문은 없음. DELETE 요청에 대한 응답이 될 수 있음
• 3xx: 경로 재지정.
• 4xx: 클라이언트 에러
  • 400 Bad Request / 401 Unauthorized / 404 Not Found / ...
• 5xx: 서버 오류

mozilla 상태코드 참고 사이트

HTTP 헤더

• 표준에 맞추면 모두에게 도움이 된다!
• 표준 요청 헤더
  • 요청에 대한 메타 데이터 개념
  • 요청 및 API 응답에 대한 추가 정보 제공
• Authorization - 문자열
  • 권한 부여: Authorization key 요청
  • 이런 표준 접근 방식을 따르면 클라이언트가 알아서 구현알 수 있다는데
  • 사실 이걸 어떻게 부여하는지는 잘 모르겠습니다!
• Date
  • 요청의 타임스탬프
• Accept - 콘텐츠 타입
  • application/xml
  • text/xml
  • application/json
• Accept-Encoding - gzip, release
  • REST 엔드 포인트는 가능한 경우 gzip과 deflate 인코딩을 항상 지원해야 한다.
  • gzip 지원은 간단함: 표준 라이브러리의 일부인 compress/gzip 패키지 사용
  • 뉴욕타임즈 오픈소스
  • 비손실 압축을 위한 인코딩,,
• 에러 리턴
  • API 사용자는 에러가 발생했을 때 여러 앤드 포인트에서 발생한 에러를 처리하는 하나의 코드를 작성할 수 있어야 하는데
  • 표준 에러 엔티티를 제공함으로써 클라이언트 또는 서버로 인한 에러가 발생할 때 클라이언트를 도와줄 수 있음
  • 마이크로소프트의 API 가이드라인 자료

API 버전 관리

• 초기부텉 고려해야하는 사항, 피할 수 있으면 피하는 것이 좋다!
• 버전 관리가 필요한 상황
  • 주요 변경 사항이 생기면 API 버전 번호를 증가시키는데 주요 변경사항은
    1. API나 API 매개 변수의 제거 또는 이름 변경
    2. API 매개 변수의 타입 변경(정수 -> 문자열)
    3. 응답 코드, 에러 코드 또는 실패 규약 변경
    4. 기존 API 의 동작 변경
• 시맨틱 버전 관리
  • 메이저 버전과 마이너 버전: 1.0 에서 1은 메이저, .0은 마이너
  • 마이너의 변경은 클라이언트가 API와 상호작용하는 데 영향을 주지 않음
• REST API의 버전 관리 형식
  • https://myserver.com/v1/helloworld :URI의 일부로 버전을 선택
  • https://myserver.com/hellowirld?api-version=1 : 쿼리 문자열의 매개 변수로 선택
  • GET https://myserver.com/helloworld api-version:2 : HTTP 헤더를 사용해 선택

객체 타입 표준화

• 클라이언트에서 객체를 쉽게 처리할 수 있도록 고려
• JSON을 사용하는 경우 기본 타입으로 날짜 개념이 없다! --> ISO 표준을 사용하는 것이 도움이 된다
• 예
  • {"date": "2021-07-08T04:52:57Z"}
  • {"date": {"kind": "U", "value": 1625719977221}

API 문서화

1. Swagger

• YAML로 작성
• 배책임님께서 이전에 소개해주신 API 문서 자동화 도구!

1. API Blueprint

• 마크다우능로 작성돼 중첩된 레이어를 다루는 것보다 좀 더 자연스럽게 문서 작성 가능

1. RAML

• RESTful API Modeling Language의 약자
• YAML로 작성

API 문서 자동화 오픈소스(https://github.com/swaggo/swag)

여러가지 참고 링크

• GO 오픈소스 참고 github
• Go REST API 프레임워크 추천

Apache Airflow에 대해 알아보자

아파치 에어플로우는 데이터 처리 파이프라인을 조율하기 위해 만든 오픈소스 도구.

Cron 같은 느낌

Apache Airflow를 사용하는 이유

• 데이터 ETL 과정을 통해 데이터를 가공하며 적재하는데
• 이때여러 개의 sequential한 로직이 존재하게 됨. (앞의 결과가 뒤 작업의 input이 되는 등)
• 위와 같은 작업이 여러개일 경우 이러한 workflow 관리도구로 airflow를 사용할 수 있으며,
• 비슷한 workflow 관리도구로 하둡 에코시스템의 우지와 같은 솔루션이 있음.

장점

• Python 기반으로 만들어 져 데이터 분석 하는 분들도 쉽게 코드를 작성할 수 있음
• Airflow 콘솔이 따로 존재해 task 관리를 서버에서하지  않아도 되고 각 작업별 시간이 나오기 때문에 bottleneck을 찾을 때에도 유용함

DAG: 방향성 비순환 그래프

• 에어플로우 상의 작업흐름은 DAG로 설계되기 때문에 이를 짤 때 어떻게 독립적으로 실행 가능한 태스크들로 나뉠 수 있을까 생각해보고
• 그 다음에 각 태스크를 그래프로 결합하여 전체적인 논리 흐름에 맞게 합칠 수 있음
• Task의 집합체이자 workflow

DAG 생성

1. default_args 정의(owner, start_date 등 정의)
2. DAG 객체 생성(dag id, schedule_interval 정의)
3. DAG 안에 Operator를 활용해 task 생성(SparkSubmitOperator, BashOperator 등)
4. Task들을 연결( >>, << 활용하거나 t1.set_upstream(t2) 또는 t2.set_downstream(t1) 도 같은 표현)

즉 DAG 객체 생성 -> Operator를 활용해 Task 작성 -> Task 연결