Hello World

Intro지난 글에서는 Producer에서 메시지를 발송하였을 때, message가 어떻게 처리되고 적재되는지 알아보았습니다. 이번 글에서는 적재된 메시지를 다시 꺼내어 처리할 때 consumer 동작 방식과 message(record) flow를 이해하는 것이 목표입니다.Consumer 구조Consumer는 크게 fetcher와 coordinator로 구성되어 있습니다.Fetcherpoll이 실행될 때, 적절한 크기의 record를 client에 return 하기 위해 Kafka Cluster로부터 record를 요청하고, client가 받기 직전에 메모리에 미리 저장하는 역할을 합니다.Fetcher 성능에 영향을 주는 Configurationfetch.min.byte ~ fetch.max.byte :..

Producer 구성요소 및 동작 방식각 구성요소 및 성능에 영향을 미치는 configuration은 아랫글을 참조합니다.AccumulatorKafka Cluster로 전송할 kafka record를 임시로 memory에 담고 있습니다. 담을 때 적용되는 주요 kafka producer configuration은 아래와 같습니다.Network Thread (io_loop)Accumulator에 쌓인 record를 Broker로 전송하는 역할을 합니다.Producer 성능에 직접적으로 영향을 주는 configurationbuffer.memory : kafka producer가 메시지를 보내기 전 메모리에 보관하는 전체 메시지 용량을 의미합니다. (Default : 32MB)max.request.size :..

0. Table Of Contents 1. 서론 2. Dead Queue의 사용 3. 문제가 생긴 offset과 error type을 기록 4. offset을 기록 하고, retention기간이 지나면 데이터를 archiving 한다. 1. 서론 현재 백엔드 팀 일부 프로젝트에서는 message를 관리하고 이를 적절한 곳에서 consume하기 위해 kafka를 사용하지만, 이에 대해서 exception handling이 잘 되고 있지 않으며 각 프로젝트 별로도 다르게 설계 중인 것으로 파악이 되었다. 따라서 kafka exception handling 어떻게 하면 잘 했다고 소문이 날지 생각해보게 되었으며, 내가 고민한 방법들을 적어보았다. 2. Dead Queue의 사용 처리가 되지 않은 message..

0. Table Of Contents 0. Table Of Contents 1. 서론 2. 이번 문제의 key는 왜 Generic인가? 3. Generic 4. Generic Type (Utility Type) 4.1. Partial 4.2. Pick 5. 최종 적용 코드 1. 서론 class 정의시, constructor를 정의하곤 했다. 일반적인 constructor를 생성시, 아래와 같은 형태를 띈다 export class ModbusConnection { constructor(modbusClient: ModbusRTU, deviceId: any, ipAddress: string, modelId: any, nestedDeviceIdList: Device[]) { this.modbusClient = ..

0. Table Of Contents 0. Table Of Contents 1. 설계에 반영되어야 하는 내용 2. Shard 및 Index 설계 3. Indice (ElasticSearch Data Modeling) 4. 데이터 시각화 1. 설계에 반영되어야 하는 내용 하나의 솔루션 데이터 뿐만 아니라 여러가지 솔루션의 데이터를 핸들링 할 수 있어야 한다. 불필요한 색인을 줄여 자원 사용량을 줄일 수 있어야 한다. 형태소 분석을 어떻게 할 것인지에 대한 전략을 작성할 수 있어야 한다. 2. Shard 및 Index 설계 설계시 생각해야 할 조건을 나열하면 아래와 같다.\ 이종간의 솔루션의 로그에 대해서 분석을 할 수 있어야 한다. ElasticSearch에서 권장하는 수준의 Shard 크기를 유지할 수 ..

0. Table Of Contents 0. Table Of Contents 1. 하고자 하였던 내용 2. 문제 분석 3. 문제 해결 4. 결론 1. 하고자 하였던 내용 ElasticSearch에 test라는 index를 생성 후, kafka consumer eachBatch 기능을 이용하여 주기적으로 document를 insert하고 있다. 이 데이터들을 이용하여 아래와 같은 효과를 내고자 한다. 실시간 데이터 모니터링 데이터 통계 ML 분석 연계 (연구가 더 필요함) 그러나 데이터를 insert 하여 모니터링 그래프를 그리는데 아래와 같은 문제가 생겼다. 2. 문제 분석 일반적인 데이터베이스의 경우, 코드 상에서 schema를 선언하거나 사전에 schema를 fix하고 작업하기 때문에 data type..

0. Table Of Content 0. Table Of Content 1. Kafka Batch data produce 도중 발생 한 문제 2. 문제를 해결하기 위해 튜닝 해야 할 파라미터에 대한 추측 3. 조치 사항 1. Kafka Batch data produce 도중 발생 한 문제 {"level":"ERROR","timestamp":"2023-04-24T08:40:10.368Z","logger":"kafkajs","message":"[Connection] Response Produce(key: 0, version: 7)","broker":"211.49.126.109:9092","clientId":"rmkr","error":"The request included a message larger tha..

0. Table Of Contents 0. Table Of Contents 1. Consumer Group이란 무엇인가 2. Consumer Group이 왜 필요한가 2.1. Consumer Failover 2.2. 다수의 Instance에서의 Offset 관리 2.2.1. 독립적인 Consumer로 구성하였을 경우 2.2.2. 같은 Consumer Group의 Instance로 구성하였을 경우 3. Consumer Group과 Partition의 관계 1. Consumer Group이란 무엇인가 Consumer Group이란, Consumer Instance를 대표하는 그룹이며 아래 그림을 참조한다. Consumer (Instance) : Broker로부터 데이터를 pull 받아 처리하는 단일 proc..

0. Table Of Contents 0. Table Of Contents 1. 발생한 이슈 사항 2. 기본적인 Kafka Topic Convention 3. 개인적으로 cut 한 생각들 1. 발생한 이슈 사항 모식도를 먼저 그려보면 아래 그림과 같다. 여러 공장에서 여러개 장비에 대한 buffer 데이터를 cluster에 전송한다. 이 때, 보낸 데이터에 대한 식별자가 필요한데 이를 topic으로 관리한다. topic convention에 대해 고민하는 도중 아래와 같은 고민이 들었다. 장비별로 topic을 생성한다 공장별로 topic을 생성한다. 이외에도 convention을 지정하기 위한 고민을 하며 들었던 생각 전체를 이 글에 적어놓고자 한다. 2. 기본적인 Kafka Topic Conventio..

0. Table Of Contents 0. Table Of Contents 1. 서론 2. 문제점 2.1. partition이 하나일 경우의 데이터 ordering test 2.2. partition이 복수개의 경우 2.2.1. kafka clustering 2.2.2. data ordering test 3. 문제점 분석 4. 해결법 4.1. partitioning key 4.2. sticky partition 1. 서론 현재 기획중인 데이터 파이프라인의 경우 아래와 같은 특성을 지니고 있다. 하나의 source가 아닌 다양한 환경의 source에서 데이터를 받는다. 현장 local server가 존재하지만, 통계 및 분석을 local server에서 지원하게되면 disk 및 성능 문제가 발생 할 수 있..