오늘은 지난 포스팅에 이어 데브원영님의 [아파치 카프카 애플리케이션 프로그래밍] 개념부터 컨슈머, 프로듀서, 커넥트, 스트림즈까지! 강의의 마지막 포스팅이 될 것 같습니다.
지난번 포스팅은 아래 링크 참고 부탁드립니다.
카프카 스트림즈
- 토픽에 적재된 데이터를 실시간으로 변환해서 다른 토픽에 적재하는 라이브러리
- 브로커 장애시 exactly once, fault tolerant system 을 가지고 있어 데이터 처리 안정성이 뛰어남
- 스트림 데이터 처리에 필요한 다양한 기능을 스트림즈 DSL, 프로세스 API 로 제공
- 소스, 싱크 토픽의
카프카 클러스터가 상이한 경우 지원 하지 않는다.(같은 클러스터 내에서만 사용 가능)
- 소스, 싱크 토픽의
스트림 내부구조
- 내부적으로 스레드 1개 이상 생성, 스레드는 1개 이상의 태스크를 가진다.
- 태스크 - 데이터 처리 최소 단위
- 3개의 파티션으로 이루어진 토픽 처리 시 내부에 3개의 태스크가 생긴다.
토폴로지
- 2개 이상의 노드들과 선으로 이루어진 집합을 뜻함
- 링형, 트리형, 선형 : 스트림즈는 트리형을 따름
프로세서와 스트림
- 소스 프로세서 : 하나 이상의 토픽에서 데이터를 가져오는 역활
- 스트림 프로세서 : 다른 프로세서가 반환한 데이터를 처리하는 역활 (변환, 분기처리 등)
- 싱크 프로세서 : 데이터를 특정 카프카 토픽으로 저장하는 역활
스트림즈 DSL 과 프로세서 API
- 스트림즈 DSL : 스트림 프로세싱에 쓰일 만한 다양한 기능들을 자체 API 로 만들어 놓음
- count, aggregation, map, flatMap 등
- 프로세서 API : DSL 에서 제공하지 않는 일부 기능 구현
스트림즈 DSL
- KStream, KTable, GlobalKTable
- 스트림즈 DSL 에서만 사용되는 개념 (프로세서 API 에선 사용불가)
KStream
- 레코드의 흐름을 표현한 것 (
메시지 키와 메시지 값으로 구성) - 컨슈머로 토픽을 구독하는 것과 동일한 선상에서 사용
- 각 파티션이 태스크에 할당
KTable
- 메시지키를 기준으로 묶어서 사용
유니크한 메시지키를 기준 최신 레코드를 사용- 각 파티션이 태스크에 할당
코파티셔닝
- KStream 과 KTable 을 조인하려면 코파티셔닝 되어 있어야 한다.
- 조인하는 2개 데이터의
파티션 개수가 동일, 파티셔닝 전략이 동일해야 한다.- 각 태스크는 KStream 의 레코드와 KTable 메시지키가 동일
- 코파티셔닝 되어 있지 않으면 조인이 불가
- TopologyException 발생
GlobalKTable
- 코파티셔닝 되지 않은 KStream 과 KTable 을 조인하고 싶을 때 사용
- GlobalKTable 로 정의된 데이터는 스트림즈의
모든 태스크에 동일하게 공유되어 사용- 데이터가 비대할 경우 태스크별 데이터를 모두 가지고 있어 무겁다.
스트림즈 DSL 중요 옵션
- 필수 옵션
- bootstrap.servers : 브로커의 host:port 1개 이상
- application.id : 스트림즈 어플리케이션을 구분하기 위한 고유한 아이디
- 선택 옵션
- default.key.serde : 레코드의 케시지 키를 직렬화, 역직렬화하는 클래스 지정 (df, byte)
- default.value.serde : 레코드의 케시지 값을 직렬화, 역직렬화하는 클래스 지정 (df, byte)
- num.stream.threads : 실행될 스레드 개수 (df. 1)
- state.dir :
상태기반 데이터 처리할 때 데이터 저장할 디렉토리 (RocksDB 활용)
스트림즈 DSL - KTable 과 KStream 을 join()
- 사용자의 이벤트 데이터를 DB에 저장하지 않고도 조인하여 스트리밍 처리할 수 있다.
스트림즈 DSL - GlobalKTable 과 KStream 을 join()
스트림즈 DSL - 윈도우 프로세싱
- 4가지 윈도우 프로세싱을 제공
- 모든 프로세싱은 메시지 키를 기준으로 취합
- 동일한 키는 동일한 파티션에 존재해야만 한다.
- 덤블링 윈도우 : 서로 겹치지 않는 윈도우를 특정 간격으로 지속적으로 처리할 때 사용
- 5분간 접속한 고객의 수를 측정하여 방문자 추이를 실시간 취합
- 호핑 윈도우 : 일정시간 간격으로 겹치는 윈도우가 존재하는 연산을 처리할 때 사용
- 동일한 키의 데이터는 서로 다른 윈도우에서 여러번 연산될 수 있다.
- 슬라이딩 윈도우 : 데이터의 정확한 시간을 바탕으로 윈도우 사이즈에 포함되는 데이터를 모두 연산에 포함
- 레코드에 포함된 타임스탬프 기반
- 세션 윈도우 : 동일 메시지 키의 데이터를 한 세션에 묶어 연산할 때 사용
- 최대 만료시간에 따라 윈도우 사이즈가 달라진다.
- 윈도우 사이즈가 가변적
- 주의 사항
커밋을 수행할 때 윈도우 사이즈가 종료되지 않아도 중간정산 데이터를 출력- 윈도우 기준으로 동일 윈도우 시간 데이터를 upsert 방식으로 처리
스트림즈 DSL - Queryable store
- KTable 은 토픽 데이터를 로컬의 RocksDB 에 Materialized View 를 만들어 사용
- 메시지 키, 값 기반으로 Key-Value Store 로 사용 가능
프로세서 API
- 토폴로지 기준으로 데이터 처리
- KStream, KTable, GlobalKTable 개념이 없다.
- 스트림즈 DSL 과 프로세서 API 를 함께 구현하여 사용할 수 있다.
- Processor : 일정 로직이 이루어진 뒤 다음 프로세스로 데이터가 넘어가지 않음
- Transformer : 일정 로직이 이루어진 뒤 다음 프로세스로 데이터가 넘길때 사용
카프카 스트림즈 VS 스파크 스트리밍
- 카프카 스트림즈 : 카프카 토픽을 input 으로 하는 경량 프로세싱 어플리케이션
- 스파크 스트리밍 : 카프카 토픽을 포함한 하둡 생태계(HDFS, hive 등)를 input 으로 복잡한 프로세싱 개발
카프카 커넥트
- 특정한 작업 형태를 템플릿으로 만들어놓은 커넥터를 실행함으로써 반복작업을 줄일 수 있다.
- 데이터 파이프라인 생성 시 반복작업을 줄이고, 효율성을 높일 수 있다.
커넥트 내부구조
- 커넥터는 태스크를 관리한다. (별도 스레드로 동작)
- 태스크는 커넥터에 종속되는 개념으로
실직적인 데이터처리를 한다.
소스 커넥터, 싱크 커넥터
- 소스 커텍터 (프로듀서 역활), 싱크 커넥터 (컨슈머 역활)
커넥터 플러그인
- jar 파일로 오픈소스 또는 직접 개발한 커넥터를 추가하여 실행
- 태스크, 커넥트 파일 포함
- 오픈소스 커넥터 : 100개가 넘는 커넥터들이 이미 공개
- 라이선스를 참고하여 사용 범위 확인 후 사용
컨버터, 트랜스폼
- 컨버팅 또는 트랜스폼 기능을 가진 플러그인
- 데이터 처리 전 스키마 변경 또는 메시지 단위로 간단한 변환
커넥트 배포 및 운영
- 단일 모드 커넥트 : 1개 프로세스만 실행
- 고가용성 보장 못함, SPOF 가 될 수 있다.
- 분산 모드 커넥트 : 2대 이상의 서버에서 클러스터 형태로 운영
- 무중단 Scale out 으로 처리량을 늘릴 수 있다.
- 커넥트 rest api : 현재 실행 중인 커넥트의 플러그인 종류, 태스크 상태, 커텍터 상태 등 조회
단일 모드 커넥트 설정
- connect-standalone.properties 파일을 수정
- 실행 시 파라미터로 커넥트 설정파일과 커넥트 설정파일을 차례로 넣어 실행
분산 모드 커넥트
- 2개 이상의 프로세스가 1개의 그룹으로 묶여서 운영
- connect-distributed.properties 로 설정
- offset.storage, config.storage, status.storage 등 토픽이 필요
- group.id 별로 다르게 설정되어야 한다.
- offset.storage, config.storage, status.storage 등 토픽이 필요
커스텀 소스 커넥터
- 소스(파일 또는 어플리케이션)로 부터 데이터를 가져와 토픽으로 넣는 역활
- 라이선스 문제나 로직이 원하는 요구사항과 맞지 않아서 직접 개발해야 하는 경우 사용
- 소스 커넥터를 빌드하여 jar 파일로 만들고 커넥트를 실행 시 플러그인으로 추가하여 사용
- connect-api 라이브러리 추가
- SourceConnector
- 태스크 실행하기 전 커넥터 설정파일을 초기화, 어떤 태스크 클래스를 사용할 것인지 정의
- validation
- SourceTask
- 데이터를 가져와서 토픽으로 데이터를 보내는 역활
토픽에서 사용하는 오프셋이 아닌 자첵적으로 사용하는 오프셋을 사용- Source 어플리케이션 또는 파일을 어디까지 읽었는지 저장하는 역활
오프셋을 통해 데이터 중복 방지
- 사용자 작성 클래스 + 참조하는 라이브러리도 함께 빌드하여 jar 로 압축
커넥터 옵션값 설정시 중요도(Importance) 지정 기준
- 커멘트 용도로 사용자로 하여금 이 옵션이 중요하다는 것을 명시적으로 표시하기 위한 문서로 사용
커스텀 싱크 커텍터
- 토픽의 데이터를 타깃 어플리케이션 또는 타깃 파일로 저장하는 역활
- 싱크 커텍터를 빌드하여 jar 파일로 만들고 커텍트를 실행 시 플러그인으로 추가하여 사용
- SinkConnector
- 실질적인 로직이 없다
- properties 를 Map 으로 가져오고, SinkTask Class 를 지정, 태스크별 컨피그 설정
- SinkTask
- start - 태스크에 사용할 properties 를 Map 으로 가져와 사용할 리소스를 초기화
- put - 레코드를 가져와서 처리하는 용도
- flush - 커밋 시점마다 데이터를 어디까지 처리했는지 오프셋 확인
- stop - 사용한 리소스 해제
- Config - AbstractConfig 상속
카프카 기술별 아키텍처 적용 방법 정리
- 카프카 커텍트
반복적인 파이프라인을 만들어야할 경우분산모드 커텍트를 설치하고 운영- 되도록 오픈소스 커넥터를 사용, 필요 시 커스텀 커넥터 개발 운영
- rest api 와 통신하여 동작할 수 있는 웹화면 개발 또는 오픈소스 활용
- 카프카 스트림즈
- 강력한 stateful(윈도우 프로세스, aggregation), stateless 프로세싱 기능, 카프카 토픽의 데이터 처리 시 선택
- 컨슈머, 프로듀서
- 커넥트와 스트림즈로 구현할 수 없거나, 단일성 파이프라인, 개발환경일 경우
카프카 기본 개념 복습
- 카프카 : 카프카 클러스터를 포함한 이벤트 스트림 프로세싱 아키텍처
- 브로커 : 카프카 클러스터의 서버 중 하나, 데이터를 저장하고 전달하는 역활(리더 파티션)
- 토픽 : 카프카 클러스터에서 데이터를 구분하는 단위, 토픽은 최소 1개 이상의 파티션 보유
- 파티션 : 토픽에서 데이터를 논리적으로 구분하는 단위
- 레코드 : 메시지를 담는 가장 작은 단위
- 오프셋 : 프로듀서가 보낸 데이터가 브로커에 저장되었을 떄 받는 고유한 번호
- 타임스탬프 : 레코드가 프로듀서에서 생성되었을 때 시간 또는 브로커에 적대된 시간
- 헤더 : 레코드의 특징을 담는 키/값 저장소
- 메시지 키 : 데이터를 구분하는 값, 파티셔너는 메시지 키를 토대로 파티션을 지정
- 메시지 값 : 실질적으로 처리하고자 하는 데이터
- ISR : 동기화가 완료된 리더, 팔로워 파티션 묶음
카프카 프로듀서
- 프로듀서 : 브로커로 데이터를 전달하는 역활을 하는 어플리케이션
- 파티셔너 : 메시지 키를 토대로 파티션을 지정하는 클래스, 커스텀 클래스를 사용하여 로직 변경 가능
- 어큐뮤레이터 : 레코드 전송 시 배치로 묶는 역활
- acks : 레코드를 클러스터로 전송 시 전달 신뢰성을 지정 (0, 1, -1(all))
- min.insync.replicas : acks = all 일 경우 최소 적재 보장 파티션 개수
- enable.idempotence : 멱등성 프로듀서로 동작하기 위해 설정
- transactional.id : 트랜잭션 프로듀서로 동작하기 위해 설정
카프카 컨슈머
- 컨슈머 : 카프카 클러스터에 저장된 레코드를 받아와서 처리하는 어플리케이션
- 컨슈머 그룹 : 동일한 역활을 하는 컨슈머 들의 묶음
- 컨슈머 랙 : 파티션에서 가장 최근의 레코드 오프셋과 컨슈머 오프셋간의 차이(버로우 통한 모니터링)
- 커밋 : 컨슈머가 레코드 처리가 완료되었을 경우 카프카 클러스터에 마지막으로 읽는 레코드의 오프셋 번호를 저장하는 작업
- 리밸런싱 : 컨슈머 그룹에서 컨슈머 개수의 변화 또는 파티션 개수의 변화로 인해 할당이 변경되는 작업
- auto.offset.rest : 컨슈머 그룹이 없을 경우 처음 읽을 오프셋의 위치를 지정하는 옵션
- isolation.level : 트랜잭션이 완료된 레코드를 읽을 것인지 판단하는 옵션
카프카 스트림즈
- 카프카 스트림즈 : 상태/비상태 기반 스트림 데이터 처리를 수행하는 어플리케이션
- 태스크 : 스트림즈 어플리케이션 내부에서 생성되어 로직을 수행하는 최소 단위
- 프로세스 : 데이터를 가져오거나, 처리하거나, 내보내는 노드
- 스트림 : 프로세서로 부터 처리된 데이터를 다른 프로세스로 전달되는 레코드
- 스트림즈 DSL : 추상화되어 스트림 프로세싱에 필요한 메서드들을 정의한 메서드들의 모음
- KStream : 레코드의 흐름, 컨슈머의 poll()과 유사
- KTable : 특정 파티션의 메시지 키를 기준으로 가장 최근의 레코드들의 묶음
- GlobalKTable : 모든 파티션의 메시지 키를 기준으로 가장 최근의 레코드들의 묶음
- 프로세서 API : 스트림즈 DSL 에서 구현할 수 없는 로직을 구현할 때 사용하는 API
- 코파티셔닝 : 동일한 파티션 개수, 동일한 파티셔닝 전략을 통해 레코드가 저장된 서로 다른 2개의 토픽
카프카 커넥트
- 카프카 커넥트 : 카프카 기반 데이터 파이프라인을 반복적으로 생성할 때 사용하는 어플리케이션
- 소스 커넥터 : 특정 파일 또는 소스 어플리케이션으로부터 토픽으로 데이터를 보내는 프로듀서
- 싱크 커넥터 : 토픽에서 특정 파일 또는 소스어 어플리케이션으로 데이터를 보내는 컨슈머
- 오픈소스 커넥터 : 소스/싱크 커넥터를 사용할 수 있도록 jar 형태로 배포하는 커넥터
- 태스크 : 커넥터에서 데이터를 처리하는 최소 로직 단위
- 단일 모드 : 디버깅, 테스트용으로 적합한 1 프로세스 단위 커넥트
- 분산 모드 : 상용환경 운영에 적합한 멀티 프로세스 단위 커넥트
마무리
그간 들은 강의 내용을 끝까지 포스팅한 적이 없었던거 같은데 처음으로 끝까지 포스팅 하는 강의가 되었네요.
앞으로 공부한 내용을 바탕으로 카프카를 다양한 프로젝트에 적용하여 사용 경험을 많이 늘려가면 좋을 것 같습니다.
그럼 이만. 🥕👋🏼🖐🏼