빅데이터/Kafka 728x90 반응형 스프링 카프카 메시지 리스너 2가지 구현 방법 2020. 7. 23. 스프링 카프카에서 메시지 리스너를 구현하는 2가지 방법이 있습니다. 첫번째는 @KafkaListener를 사용하는것이고 두번째는 listenerContainer를 Bean등록하는 것입니다. @KafkaListener로 구현 가장 간단한 방법입니다. @KafkaListener를 통해 선언할 경우 파라미터를 오버로딩해서 알맞는 listenerContainer를 자동으로 주입합니다. package com.example; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBoot.. 카프카 컨슈머 파티셔너 종류 및 정리(2.5.0 기준) 2020. 7. 23. 카프카 컨슈머는 토픽의 파티션과 매칭하여 레코드를 가져옵니다. 파티션을 매칭하는 기준은 컨슈머 파티션 어사이너의 기준을 따릅니다. 컨슈머 파티션어사이너 인터페이스는 아래 링크에서 확인할 수 있습니다. https://kafka.apache.org/25/javadoc/org/apache/kafka/clients/consumer/ConsumerPartitionAssignor.html ConsumerPartitionAssignor (kafka 2.5.0 API) kafka.apache.org RangeAssignor 기본 설정되는 파티션 어사이너입니다. 토픽의 파티션을 숫자기준으로 나열하고, 컨슈머의 이름을 사전순으로 나열한 뒤에 배정하는 정확히 반으로 나누어 배정합니다. 만약 딱 반으로 안나뉘어지는 홀수개의 .. 카프카 프로듀서 파티셔너 종류 및 정리(2.5.0 기준) 2020. 7. 23. 카프카 프로듀서는 레코드를 전송하기 위해 파티셔너를 제공합니다. 파티셔너 종류와 각 파티셔너에 대한 설명을 정리해보았습니다. 카프카 버젼이 올라감에 따라 파티셔너의 종류도 달라졌는데 여기서는 2.5.0 버젼의 파티셔너를 정리해보겠습니다. 프로듀서 파티셔너 인터페이스에 대한 설명은 아래 링크에서 확인할 수 있습니다. https://kafka.apache.org/25/javadoc/?org/apache/kafka/clients/producer/Partitioner.html kafka 2.5.0 API kafka.apache.org 메시지 키가 있을 때 메시지 키가 있을 경우에는 파티셔너의 종류와 관계없이 동일하게 동작합니다. 메시지 키의 해쉬값을 구해서 해당 해쉬값과 파티션을 매칭하여 적재합니다. 이로 인해.. 컨슈머 스레드가 많다고 처리량이 높을까? 아닐까? 컨텍스트 스위칭으로 인한 예외 상황 2020. 7. 22. 오늘 카프카 컨슈머를 입수하던 도중 매우 신기한 현상을 발견해서 기록하게 되었습니다. 컨슈머스레드를 여러개로 HDFS에 적재하고 있었는데, 일부 파티션의 데이터가 0이 되어 일부 컨슈머 스레드가 쉬는동안 총 처리량이 늘어난 것입니다. 그래프를 보면서 왜 그렇게 되었는지 확인해보겠습니다. 입수 상태 - 24 core, 48G memory 물리장비 - 1 JVM application - 20 consumer thread - sync offset commit - save at Hadoop hdfs(append) 입수 관련 그래프 컨슈머 랙은 카프카 버로우를 사용하여 수집중에 있습니다. 랙 추이를 보면 알겠지만 파티션당 데이터 양의 차이가 일부 발생하였습니다. 10개 파티션은 18:20 부근에 끝났고 나머지 1.. kafka connector distributed 모드로 fileSourceConnector 실행 2020. 7. 21. 저번 포스트에 이어 distributed로 동작하는 kafka connector를 실습해보겠습니다. https://blog.voidmainvoid.net/356 Kafka file source connector standalone 모드 실행 Kafka에는 커낵터가 있습니다. 다양한 커낵터 클래스를 사용하여 컨슈머나 프로듀서 작성 없이 source로 부터 데이터를 카프카로 보내거나 받을 수 있습니다. 오늘은 FileStreamSource를 사용해서 file을 blog.voidmainvoid.net 준비물 - 카프카 바이너리 - 실행중인 카프카 클러스터 distributed vs standalone The difference is in the class which is started and the confi.. Kafka file source connector standalone 모드 실행 2020. 7. 21. Kafka에는 커낵터가 있습니다. 다양한 커낵터 클래스를 사용하여 컨슈머나 프로듀서 작성 없이 source로 부터 데이터를 카프카로 보내거나 받을 수 있습니다. 오늘은 FileStreamSource를 사용해서 file을 카프카로 보내보겠습니다. 준비물 - 카프카 바이너리파일 - 카프카 클러스터 Standalone 모드 준비 Standalone모드란 1개의 프로세스로 동작하는 애플리케이션을 뜻합니다. 커낵터는 distribute모드와 standalone모드를 지원합니다. 실제 환경에서는 distribute모드를 사용하는 것을 권장합니다. standalone모드를 실행하기 위해 아래와 같이 준비합니다. connect-file-source.properties name=local-file-source conne.. Kafka-client 사용시 Failed to load class "org.slf4j.impl.StaticLoggerBinder" 에러 해결 방법 2020. 7. 13. kafka-client를 java application에 추가하여 실행하면 아래와 같은 오류와 함께 log가 정상적으로 찍히지 않습니다. SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder". SLF4J: Defaulting to no-operation (NOP) logger implementation SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details. 위 에러는 slf4j의 디펜던시 이슈로 StaticLoggerBinder가 없어서 그런것인데요. 아래와 같이 그래들에 디펜던시를 추가하면 로그를 확인하실 수 있습니다. dependencies { co.. Kafka ConsumerRecord의 timestamp는 0.10.0.0 이후부터 사용가능합니다. 2020. 7. 8. 카프카의 ConsumerRecord를 살펴보면 메시지 키와 메시지 값 이외에 추가로 timestamp가 있는 것을 확인할 수 있습니다. timestamp는 Kafka 프로듀서가 ProducerRecord를 생성하면서 timestamp 값을 Unix time으로 입력합니다. 만약 다른 timestamp를 넣고 싶다면 직접 설정하여 넣을수도 있습니다. 해당 건은 KAFKA-3025에서 논의했던 내용으로 자세한 내용은 해당 jira에서 확인할 수 있습니다. https://issues.apache.org/jira/browse/KAFKA-3025 [KAFKA-3025] KIP-31&KIP-32 (part 1): Add timestamp field to message, configs, and Producer/Con.. telegraf사용시 kafka로 데이터 json형태로 보내는 방법 2020. 7. 8. telegraf는 metric수집에 최적화되어있습니다. 다만, 그대로 사용하게 되면 influxdb에서 사용하는 데이터형태(csv와 유사)로 데이터가 전송되는데요. kafka에서 데이터를 잘 활용하기 위해 json을 사용한다면 consumer에서도 스키마에 대한 걱정없이(?) 잘 사용할 수 있습니다. telegraf에서 data형태를 json형태로 바꾸는 예제를 아래에 설명드리도록 하겠습니다. telegraf에 대한 설명은 아래 링크를 확인해주세요. https://blog.voidmainvoid.net/260 Fluentd vs Telegraf 차이점 알아보기 Telegraf와 fluentd는 아주 유사해보인다. 둘다 configuration파일 기반으로 작동하며 plugin을 통해 개발자가 custom.. 카프카의 토픽 데이터를 REST api로 주고받자 - Kafka rest proxy 사용 2020. 6. 25. confluent에서는 rest proxy라고 불리는 카프카 클러스터를 위한 RESTful interface application을 오픈소스로 제공하고 있습니다. 기존에 Kafka connect, Kafka client로 데이터를 전달하는 것과는 사뭇 다르게 REST api를 사용한다는점이 독특한데요. 직접 코드를 짜지 않고 범용적으로 사용되는 http을 사용해서 데이터를 넣고 뺄 수 있다는 점이 독특합니다. 오늘은 rest proxy를 local에 설치하고 실행해보겠습니다. 준비물 - local kafka cluster - git - terminal - postman 다운로드 및 실행 rest proxy를 사용하기 위해서는 rest proxy가 포함된 confluent의 community package.. AWS MSK 사용시 인스턴스 유형별 최대 토픽 개수 2020. 6. 24. AWS MSK는 카프카를 SaaS형태로 사용가능한 AWS서비스중 하나입니다. 완전관리형 SaaS는 아니지만 다양한 옵션을 지정할 수 있으며, 추가적인 모니터링 도구를 제공해주기 때문에 아주 편리하게 사용할 수 있습니다. MSK에서는 총 7개의 인스턴스유형을 제공하는데, 각 인스턴스 유형별로 생성할 수 있는 토픽의 개수를 살펴보도록 하겠습니다. MSK Bestcase 문서에 따르면 적정크기의 클러스터에 대한 내용이 나와 있습니다. 여기서는 MSK클러스터를 생성할 때 브로커 노드의 유형 및 수에 대한 상세한 내용이 나와 있는데 그 중 브로커당 파티션수에 대한 내용이 아래와 같이 정리되어 있습니다. 위 그림에서 보는 것 처럼 t3.small이 가장 작고 m5.24xlarge가 가장 큰 파티션 개수를 가질 수 .. 아파치 카프카 입문과 활용 강의자료 슬라이드 2020. 6. 23. 강의자료 : https://www.slideshare.net/WonyoungChoi2/ss-236018398 강의코드 : https://github.com/AndersonChoi/tacademy-kafka 자바 멀티스레드 카프카 컨슈머 애플리케이션 구현 코드 2020. 6. 9. 이번 포스트에서는 자바 멀티스레드 카프카 컨슈머 애플리케이션을 구현해보도록 하겠습니다. 준비물 - Intellij 또는 eclipse - gradle project - jdk 1.8 - 약 10분 구현방법 이전 포스트(카프카 컨슈머 멀티쓰레드 애플리케이션 예제코드(for scala) 바로가기)에서 scala로 구현한 바가 있습니다. 이번에는 자바로 구현하고, consumer.wakeup()을 사용해서 consumer를 안전하게 해제시키는 것을 아래 코드에서 구현해보겠습니다. wakeup() 메서드는 java shutdown hook을 통해 호출하여 안전하게 종료되도록 설정해보겠습니다. 기존 구현방식 처럼 newCachedThreadPool 메서드를 사용해서 multiple thread를 생성해보겠습니다... Kafka-client client.dns.lookup 옵션 정리 2020. 4. 13. Kafka-client 2.1.0 이후 버젼에서는 client.dns.lookup옵션을 사용하여 dns관련 설정을 사용할 수 있습니다. 오늘 포스팅에서는 해당 옵션이 어떤 역할을 하고 어떻게 동작하는지 알아보도록 하겠습니다. 먼저, apache kafka document의 설명을 보도록 하겠습니다. client.dns.lookup document client.dns.lookup: Controls how the client uses DNS lookups. If set to use_all_dns_ips then, when the lookup returns multiple IP addresses for a hostname, they will all be attempted to connect to before .. 카프카 클러스터 클러스터ip DNS 연동방법. use_all_dns_ips 사용(in AWS, route53) 2020. 3. 20. Kafka-client(consumer, producer)를 사용하기 위해서는 다양한 설정이 필요하지만 카프카 브로커와 통신하기 위해서는 bootstrap.servers 옵션은 반드시 필요한 옵션중 하나입니다. Bootstrap.servers 이 옵션은 카프카 클러스터에 연결하기 위해 클라이언트가 사용하는 브로커들의 host:port 목록을 설정해야 합니다. 특이한점은 모든 브로커의 host와 port를 적지 않아도 된다는 점입니다. 왜냐면 최초로 연결된 하나의 broker의 host:port로 부터 통신을 위한 정보를 가져오기 때문입니다. Route 53에서 kafka cluster DNS 설정하기 route53은 aws에서 제공하는 DNS 웹서비스 입니다. aws route53에 이미 등록되어 있는 .. 이전 1 2 3 4 5 6 7 8 다음 728x90
