# kafka
Kafka - 신뢰성 있는 카프카 애플리케이션을 만드는 3가지 방법
신뢰성 있는 카프카 애플리케이션을 만드는 3가지 방법 이 글은 최원영님의 kakao tech meet 발표를 보고 공부한 내용을 정리한 것 입니다. Kafka의 사용 이유 EDA(Event Driven Architecture)와 Stram Data Pipeline 에서 중요한 역할을 수행 스트림 이벤트를 다뤄 실시간 데이터 처리 타임스탬프, 파티션, 메시지 키와 같은 기능들을 활용해 시간 단위 이벤트 데이터를 다룸 Meassage Reliability 
n-1개의 복제계수에 대해서
최근 n-1개의 복제계수로 토픽을 만들면 어떻게 메세지가 복제가될까라는 고민을 많이 했습니다. (n = 클러스터에서 브로커의 수) 파티션과 복제 파티션 파티션이란 메시지들이 저장되는 곳입니다. 하나의 토픽은 여러 개의 파티션으로 나뉘어져 있으며, 각각의 메시지는 하나의 파티션에 저장됩니다. 복제 계수 복제 계수는 각 파티션 데이터가 클러스터 내에서 몇 번 복사되어 유지되는지를 결정하는 값입니다. 만약 복제 계수를 2로 설정한다면, 각 파티션 데이터는 두 개의 브로커(서버)에 저장됩니다. 이 때 한 브로커가 리더 역할을 하고 다른 하나가 팔로워 역할을 합니다. 복제계수를 설정하는 목적 데이터 안전성 시스템 가용성을 확보 만약 어떤 한 서버
Apache Kafka 개요
배경 요즘의 개인적인 소프트웨어 엔지니어링의 화두는 서버리스와 함수형 프로그래밍인 듯 하다. 그러나 이런 기술이 나오게 된 배경은 대부분의 IT 서비스가 빅데이터 환경에서 제공되기 때문이라고 생각한다. 즉, 대량의 데이터를 주어진 시간 안에 처리해야 한다. 그러기 위해선 첫 번째로 반드시 병렬 처리를 사용해야 한다. 그리고 확장 가능한 시스템을 사용해야 한다. 대량의 데이터를 처리할 수 있어야 할 필요성은 있지만 그것이 어느 시기에 얼마만큼 지속될 지는 알 수 있기 때문이다. 그리고 이런 시스템은 클라우드 환경에서 제공하기 적합하다. 그러나 사용이 어렵기 때문에 높은 추상화를 제공하는 서버리스의 선호도가 올라가고 있는 것이다. 그리고 카프카 또한 이런 시대적 배경 속에서 태어났으며, 비동기적으로 이벤스 드라이븐 아키텍처를 사용하여 Pub/Sub 구조의 메시지 브로커 역할을 수행할 수 있다. 아키텍처 
Kafka commands(정리중....)
kafka topic 리스트 출력 topic 생성 --bootstrap-server : 명령어를 보내는 broker(kafka server)의 주소 및 포트 번호 --partitions : 전송하는 데이터를 몇개로 분할하는지에 대한 값 --replication-factor : 분할된 partitions를 몇개로 복제하는지에 대한 값 >[!info] >--partitions 와 --replication-factor의 값은 broker 수보다 적어야 한다. ex) 위 명령어로 topic을 생성했을 때 broker에 생긴 topic의 상태 , topic의 데이터는 1,2,3,4라고 가정한다. | broker | replication-factor 1 | replication-factor 2 | | ------ | -------------------- | -------------------- | | kafka1 | partition1 (1,2)
MSA - 소모임 project
기획의도 각자의 취미 모임을 가입하여 여가 활동을 즐기는 어플 소모임을 보고 카피하여 만든 프로젝트이다. 하나의 서버에 문제가 생겨도 다른 서버에 영향을 주지 않기 위해 MSA 기반으로 프로젝트를 진행 할 예정이다. Netflix Eureka Server 로 Server 들을 관리하며 load Balancer 를 사용할 예정이다. Eureka 안에서 openfeign 으로 Server 끼리의 소통을 진행할 예정이다.(다음 주 공부하는 kafka 로 고도화 예정) 프론트에선 모든 서비스로 request 를 뿌려주는 netty 기반 Server로 GateWay 역할을 구축할 예정. (non-blocking) openfeign 을 통해 소통 하던 부분을 구축하고, 후에
카프카 9
제 3의 그룹이 토픽이 필요할때 만들고 더 이상 니즈가 없으면 컨슈머 삭제하고 토픽도 삭제. 레코드 안에 타임 스탬프를 기록하여 병렬 처리 하여도 타임 스탬프 순서로 처리. 파티션은 기본적으로 3개 사용 하지만 컨슈머 랙에 따라 파티션을 늘림. 파티션을 늘리는 일은 큰 일이다. 리밸런스에 영향을 주고 줄이는 것도 안된다. 
카프카 8
컨슈머 랙 컨슈머 마지막 커밋 오프셋과 토픽의 마지막 오프셋 차이. 컨슈머 랙을 모니터링 하는것은 중료. 컨슈머 랙이 크다는 것은 서비스 처리가 느리다는 의미. 컨슈머 랙은 컨슈머의 상태를 나타내는 지표. 컨슈머 랙의 최대값은 컨슈머 인스턴스를 통해 직접 확인할 수 있음. consumer.metrics()를 통해 확인할 수 있는 지표. record-lag-max : 토픽의 파티션 중 최대 랙 fetch-size-avg : 1번 polling 하여 가져올 때 레코드 byte 평균. fetch-rate : 1초 동안 레코드 가져오는 회수 컨슈머 랙을 직접 수집할 수 있지만 좋은 방법은 아니다. 인스턴스가 죽어 버리면 확인할 수 없다. 
카프카 7
컨슈머 쓰레드 전략 1 : 1 프로세스 + 1 쓰레드(컨슈머) 간략한 코드 프로세스 단위 실행/종료 다수의 컨슈머 실행 필요시 다수의 프로세스 실행 필요 2 : 1 프로세스 + n 쓰레드(동일 컨슈머 그룹) 복잡한 코드 쓰레드 단위 실행/종료 스레드간 간섭 주의(세마포어, 데드락등) 다수 컨슈머 실행시 다수 쓰레드 실행 가능 3 : 1 프로세스 + n 쓰레드(다수 컨슈머 그룹) 복잡한 코드 컨슈머 그룹별 쓰레드 개수 조절 주의 
카프카 6
리밸런스 컨슈머 그룹의 파티션 소유권이 변경될 때 일어나는 현상. 컨슈머와 파티션 할당 변경. 리밸런스를 하는 동안 일시적으로 메시지를 가져올 수 없으며 리밸런스 발생시 데이터 유실/중복 발생 가능성 있음. commitSync() 또는 추가적인 방법(unique key)으로 데이터 유실/중복 방지. 추가적인 방법으로 유니크 키같은 것으로 유실을 방지할 수 있음 ? 리밸런스 발생 시점 consumer.close() 호출시 또는 consumer의 세션이 끊어졌을때 > 보통 Headbeat * 3 을 세션 시간으로
카프카 5
컨슈머 데이터를 가져가는 polling 주체. 커밋을 통해 읽은 컨슈머 offset을 카프카에 기록. 데이터 저장 위치. FileSystem(.csv, .log, .tsv) Object Storage(S3, Minio) Hadoop(Hdfs, Hive) RDBMS(Oracle, Mysql) NoSql(MongoDB, CouchDB) 기타 다양한 저장소들(Elasticsearch, influxDB) Simple Consumer 프로듀서에서 레코드 주입 컨슈머에서 레코드를 가져옴 Auto commit Consumer enable.auto.commit=true : 일정 간격(auto.commit.interval.ms), poll() 메서드 호출시 자동 commit commit 관련 코드를 작성할 필요 없기에 편리함. 속도가 가장 빠름. 그러나 중복 또는 유실이 발생할 수 있음(중복/유실을 허용하지 않는 상황에서는 사용하면 안됨) 일부
카프카 4
Simple Producer Key, Value Producer key, value가 제대로 들어 오는지 확인하기 위해서 콘솔 컨슈머를 다음 옵션과 함께 실행 key 값을 넣지 않는 방식으로 레코드 전달 할 경우 레코드 키 카프카가 전달하는 메시지는 레코드라고 한다. 레코드 키는 메시지를 구분하는 구분자 역할을 한다. 동일 키는 동일 파티션에 적재한다. 디폴트 파티셔너의 경우 키를 해쉬값으로 하여 동일한 파티션으로 저장 하도록 수행. 순서를 보장하므로, 상태 머신으로 사용 가능하다. 역할에 따라 컨슈머를 할당 및 적용할 수 있다. 
카프카 3
그룹 컨슈머에서 그룹을 지정하여 메시지를 polling 할 경우 다음과 같이 동작한다. 지정한 그룹을 추가한다. 메시지를 읽은 offset을 기억하여 offset 부터 읽기 시작한다. N개의 파티션에 기록된 메시지를 병렬적으로 가져온다(입력된 메시지 순서가 보장되지 않을 수 있다) 생성된 그룹 확인 그룹 상태 확인 컨슈머 오프셋 : 컨슈머가 현재까지 읽은 오프셋 로그 엔드 오프셋 : 토픽의 마지막 오프셋 컨슈머 랙 : 컨슈머 오프셋과 로그 엔드 오프셋의 차이. 이 차이가 크다는 뜻은 컨슈머가 느리게 처리하고 있음을 의미. 컨슈머 오프셋을 리셋하여 컨슈머가 처음부터 메시지를 읽도록 처리 하고 싶을 수 도 있다. 특정 파티션의 offset을 특정 값으로 초기화 하는 것도 가
카프카 2
카프카 설치 카프카는 설치 과정이 없는 포터블 패키지형 프로그램이다. 따라서 카프카 사이트에서 압축 파일 다운로드 받은 후 적당한 디렉터리에 압축 해제하면 사용 가능하다. bin 디렉터리에 커맨드형 프로그램들이 있고 config 디렉터리에 설정 파일 있다. server.properties 카프카 서버 실행 카프카 명령을 실행하기 위해서는 카프카 서버가 실행되어 있어야 한다. 주키퍼 실행 : 2.X 버전의 카프카 서버는 주키퍼가 미리 실행되어 있어야 한다. 카프카 서버 실행. 카프카 기본 테스트 토픽 생성 생성된 토픽에 콘솔 프로듀서로 메시지 입력 콘솔 컨슈머를 사용하여 프로듀서가 입력한 메시지를 처음부터 받아오기 콘솔 컨슈머를 종료 후 다시 같은 명령을 실행하여도
카프카
Before 카프카 엔드투엔드(end-to-end) 연결 방식의 아키텍쳐. 데이터 연동의 복잡성 증가(HW, OS, System, ...). 각기 다른 데이터 파이프라인 연결 구조. 확장에 엉청난 노력 필요. 모든 시스템으로 데이터를 전송 실시간 처리도 가능한 것 데이터가 갑자기 많아지더라도 확장이 용이한 시스템이 필요. After 카프카 프로듀서/컨슈머 분리 메시지 데이터를 여러 컨슈머에게 허용 높은 처리량을 위한 메시지 최적화 스케일 아웃 가능 관련 생태계 제공 카프카 브로커 실행된 카프카 어플리케이션 서버(프로세스) 중 1대. n 개 이상 사용할 수 있지만 그렇게 운영하는 경우는 없음. 3대 이상의 브로커로 클러스터 구성 가능. 카프카 클러스터에서 1대는 컨트롤러 기능 수행. 각 브로커에게 담당파티션 할당 수행. 브로커 정상 동작 모니터링 관리. 주키퍼와 연동(~2.5.0 버전) 필요. 주키퍼의 역할 : 메타 데이터(브
실시간 뉴스 데이터 처리로 Kafka 사용기 하(下)
2023.08 지난 포스팅에서는 네이버 뉴스에서 데이터를 크롤링하는 부분까지 구현하였습니다. 이번에는 크롤링한 데이터를 전달하여 Broker에서 받은 데이터를 S3에 적재하는 과정까지 진행하려고 합니다. Local에서 Kafka 실행하기 Local에서 데이터 전달하기(Producer) 먼저 이전에 작성했었던 Producer.py에 Kafka 라이브러리를 사용할 것입니다. today-news라는 토픽을 해당 broker 서버에 전달할 예정입니다. 우선, Local에서 진행하기 때문에 broker 서버는 localhost:9092로 설정해놓습니다. > * 왜 포트번호를 9092로 설정해놓은 이유가 무엇일까요?* 9092포트는 kafka 내부 설정 파일인 server.properties을 확인해보면 “listeners=PLAINTEXT://:9092”로 설정이 되어있을 것입니다. default 일뿐이고 config파일에서 설정 가능합니다.
실시간 뉴스 데이터 처리로 Kafka 사용기 상(上)
Kafka를 공부하면서 배웠던 내용에 대해 포스팅해보려고 합니다. 겸사로 배웠던 지식을 활용해 실시간으로 뉴스 데이터를 전달하는 작은 프로젝트를 해보려고 합니다. Kafka? 우선, Kafka는 실시간으로 들어오는 데이터를 처리할 수 있는 분산 메세지 큐입니다. 이외에도 C기반인 Red-Panda 또한 주목을 받고 있는 분산 메세지 큐입니다. > 여기서 분산 메세지 큐란? 분산 메세지 큐를 이해하기 전에 Event Driven Architecture;EDA에 대해 짚고 넘어가야 합니다. EDA는 Pub-Sub 구조를 갖고 있습니다. Pub-Sub 구조란, 이벤트를 감지하고 수집, 전송을 담당하는 Pub이 Event Channel을 통해 Sub에게 전달하는 구조를 말합니다. 아무튼 EDA를 만들기 위해 가운데 Event Channel이 중요한 역할을 하는데 Event Channel에 대한 의존
Kafka cluster 구성
Zookeeper 설치 kafka를 동작시키기 위해서는 zookeeper가 필수적이다. zookeeper를 다음 link를 보고 설치를 해준다. https://velog.io/@dldydrhkd/Zookeeper-cluster-%EA%B5%AC%EC%84%B1 Kafka 설치 위 링크대로 zookeeper의 설치가 끝나고 zookeeper를 실행 시켰다면 kafka를 설치해보자. 우선 kafka 폴더를 생성한다. 아래 링크를 들어가 원하는 버전을 설치하고 압축을 해제한다. https://kafka.apache.org/downloads 경로 설정 kafka의 경로 설정을 위해 bashrc 파일을 연다 다음과 같이 경로를 추가한다. 수정된 사항을 적용 시킨다. kafka 환경 설정 cluster 구성을 위한 kafka 설정을 해준다. config 폴더에 들어간다. server.prope
크롤링 데이터를 kafka를 통해 Elasticsearch에 저장하기
0. 고민 >1. 카테고리별 elasticsearch의 인덱스로 분리해서 저장하기 초기에는 kafka connect를 이용해서 kafka-elasticsearch 파이프라인을 구축했기 때문에 SMT를 이용해서 필드의 값을 인덱스로 설정하려고했다. 하지만 아래 문제의 parent-child관계는 SMT방식으로 한계가 있다고 판단해서 kafka - logstash - elasticsearch 로 파이프라인을 변경했다. output과정에서 index_name필드를 그대로 인덱스로 지정하면 되기 때문에 훨씬 간단하게 해결가능했다. >2. 상품 상세 조회시 가격 추이 그래프를 제공하기 위해서 제목, 링크 등의 기본 정보와 시간별 가격 정보를 분리해서 저장시켜야 했다. 함께 저장할 시 최대 5분마다 업데이트가 진행되는데 가격외에 고정적인 정보까지 같이 접근되기 때문에 따로 저장하기로 했다. 그래서 찾은 방법은 기본 정보와 가격 정보를 parent-child관계로
토스증권 실시간 시세 적용기
토스를 다니는 건 아닙니다 썸네일이... 회사에서 곧 진행할 프로젝트 관련 참고 영상을 보며... 정리한 내용 그런데 이거 너무 어려운데...? 토스ㅣSLASH 22 - 토스증권 실시간 시세 적용기 영상을 보며 정리한 내용입니다 실시간 시세 적용기 주문호가 : 거래를 위해 서로 사거나 팔고 싶은 가격을 부르는 것 → 일치하는 상대 주문이 있을 경우 거래가 성사! > 성사된 거래 가격들이 쌓인 것 = 주식 차트의 체결 시세 > 가장 최근의 체결 가격 = 현재가 증권투자자들은 자신에게 유리한 조건으로 거래를 하기 위해 가장 최신의 정보를 근거로 호가를 결정하길 원한다 이 때 가장 기본이 되는 정보 = 시세와 주문 호가 >사용자들 → **최신의 시세와 호가 정보를 제공하는 MTS
Kafka란?
Kafka란? Kafka는 Pub-Sub 모델의 메시지 큐이다. 분산환경에 특화되어있는 특징을 가지고 있다. 구성요소 Event Event는, kafka에서 producer와 Consumer가 데이터를 주고 받는 단위이다. Producer Producer는 kafka에 이벤트를 게시하는 클라이언트 애플리케이션을 의미한다. Consumer Consumer는 이러한 Topic을 구독하고 이로부터 얻어낸 이벤트를 처리하는 클라이언트 애플리케이션이다. Topic 이벤트가 쓰이는 곳이다. Producer는 이 Topic에 이벤트를 게시한다. 그리고 Consumer는 Topic으로부터 이벤트를 가져와 처리한다. Topic은 파일시스템의 폴더와 유사하며, 이벤트는 폴더안의 파일과 유사하다. Topic에 저장된 이벤트는 필요한 만큼 다시 읽을 수 있다. partition Topic는 여러 Broker에 분산되어 저장되며, 이렇게 분산된 Topic을 P