빅데이터/Kafka 728x90 반응형 카프카 스트림즈 All stream threads have died. 오류 해결 방안 2021. 5. 27. 카프카 스트림즈 사용시 토폴로지 내부 에러로 인해 아래와 같은 오류가 발생할 수 있다. All stream threads have died. The instance will be in error state and should be closed. 이에 대응하기 위해 2.8.0에서 추가된 setUncaughtExceptionHandler메서드를 사용하면 스레드를 재시작하여 데이터를 지속 처리할 수 있다. streams.setUncaughtExceptionHandler((exception) -> StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.REPLACE_THREAD); 2.8.0 이전에는 setUncaughtExceptionHandler(fin.. ProducerRecord에 파티션 번호를 지정하면 어떻게 동작할까? 2021. 5. 5. ProducerRecord는 다양한 파라미터를 받고 그 중 파티션번호를 직접 넣어서 사용하는 경우도 있습니다. 이 때 파티션 번호를 넣으면 어떻게 동작할까요? 정답은 KafkaProducer private partition()메서드를 보면 확인할 수 있습니다. /** * computes partition for given record. * if the record has partition returns the value otherwise * calls configured partitioner class to compute the partition. */ private int partition(ProducerRecord record, byte[] serializedKey, byte[] serializedV.. 카프카를 이벤트 소싱, CQRS로 사용할 수 있을까? 2021. 5. 3. 우선 이벤트 소싱과 CQRS에 대한 개념부터 짚고 넘어가겠습니다. 카프카에 대한 개념은 이 동영상을 참고해주세요. 이벤트 소싱 패턴 이벤트 소싱은 배치성 데이터 저장소에 현재 상태만 저장하는 것이 아니라 이벤트 브로커(전용 저장소)에 발생된 모든 이벤트 기록(레코드)들을 기록하는 것을 뜻합니다. 이를 통해 데이터 모델과 비즈니스 도메인을 동기화할 필요가 없어지고 성능, 확장성, 응답성이 향상되어 도메인 태스크를 간소화 할수 있습니다. 기존에는 배치성 데이터 저장소(오라클, mysql 등)에 서비스에 들어오는 명령(또는 이벤트)이 발생할 때 마다 항상 동기화 했었습니다. 간단한 구조와 적은 데이터 접근만 일어날 때는 큰 이슈가 없지만, 데이터과 이벤트가 많아질 수록 데이터에 대한 CRUD와 같은 접근으로 .. confluent-kafka-go 컨슈머를 구현하는 5가지 방법 2021. 4. 30. confluent-kafka-go는 컨플루언트에서 개발하고 유지보수하는 golang기반 카프카 클라이언트입니다. github.com/confluentinc/confluent-kafka-go confluentinc/confluent-kafka-go Confluent's Apache Kafka Golang client. Contribute to confluentinc/confluent-kafka-go development by creating an account on GitHub. github.com 위 라이브러리를 사용해서 컨슈머 5가지 패턴을 구현해보겠습니다. 컨슈머 구현 사전 작업 코드(컨슈머 초기화) 이 작업은 컨슈머를 이용하기 위한 가장 기본적인 작업입니다. 기본 옵션은 enable.auto.com.. 레디슈 큐(queue), 레디스 스트림(streams), 레디스 펍섭(pub/sub) 그리고 카프카와 비교 2021. 4. 25. 레디스란? 레디스는 오픈소스 인 메모리 데이터 구조 저장소로서 데이터베이스, 캐시, 메시지 브로커로 역할을 수행한다. 레디스는 String, hash, lists, sets, sorted sets, bitmaps, streams등을 지원한다. 레디스 큐란? 레디스 큐는 레디스의 자료구조 중 List를 이용하여 Queue를 구현한 것이다. 큐는 FIFO(First In First Out)구조로 먼저 들어온 데이터가 먼저 처리되는 것이다. LIST자료구조의 LPUSH, RPOP(또는 RPUSH, LPOP)을 사용하여 구현할 수 있다. 이외에도 BLPOP을 사용하여 블락킹(blocking) pop을 수행할 수 있다. 지정한 시간만큼 기다리고 값이 들어오면 LPOP을 수행하는 것이다. $ lpush my-que.. 카프카 스트림즈의 commit.interval.ms옵션 2021. 4. 15. 카프카 스트림즈에서 commit.interval.ms 옵션은 프로세서의 frequency 즉, 반복을 얼마만큼 인터벌 간격으로 할 것인가에 대한 것이다. The frequency in milliseconds with which to save the position of the processor. (Note, if processing.guarantee is set to exactly_once, the default value is 100, otherwise the default value is 30000. 예를 들어 KTable로 선언된 변수를 toStream으로 변경하는 구문이 있다고 가정하자. 이 때 30000이면 30초 인터벌을 가지고 변경을 수행한다. 만약 짧게 가져가고 싶다면 0이나 아주 작은 숫.. 초~중급자를 위한 [아파치 카프카 애플리케이션 개발] 서적을 출간하였습니다. 2021. 4. 11. 2020년 6월 부터 집필을 시작하여 6개월간의 원고 집필과 3개월간의 편집 끝에! 「아파치 카프카 애플리케이션 프로그래밍 with 자바」를 출간하게 되었습니다. 출간일은 4월 14일로 화요일부터 구매 순서대로 배송이 시작됩니다. - 예스24 : www.yes24.com/Product/Goods/99122... - 교보문고 : www.kyobobook.co.kr/product/detai... - 알라딘 : www.aladin.co.kr/shop/wproduct.aspx?... 책을 집필하게 된 계기 사실 개발 블로그와 개발 유튜브를 진행하면서 딱히 책을 쓸 생각은 없었습니다. 책을 쓰는데 들이는 시간과 노력이 생각보다 크다는 것을 알고 있었기 때문입니다. 하지만 제 컨텐츠들을 봐주시고 '카프카를 주제로'.. 카프카 스트림즈 suppress() 사용할 때 ClassCastException 문제 2021. 4. 6. 카프카 스트림즈에는 suppress()메서드가 있다. 이 메서드를 통해 일정 윈도우 간격 안에 데이터를 모아서 처리할 수 있는데, 이슈가 있다. 값을 모아서 state store(rocksDB)에 저장할 때 제대로 deserialize를 하지못하는것이다. java.lang.ClassCastException: org.apache.kafka.streams.kstream.Windowed cannot be cast to java.lang.String at org.apache.kafka.common.serialization.StringSerializer.serialize(StringSerializer.java:28) at org.apache.kafka.streams.kstream.internals.suppress.. Kafka streams 에러 : Size of data received by LongDeserializer is not 8 2021. 4. 6. 이 에러는 카프카 스트림즈 내부에서 null 메시지 값에 대한 처리를 제대로 못해서 나오는 이슈이다. 그러므로 KStream에서 filter로 null 메시지 값을 삭제하여 처리하면 된다. KTable KT01 = myStream .filter((k, v) -> v != null) // null value 제거 ... 카프카 스트림즈 애플리케이션 초기화 명령 2021. 3. 31. 카프카 스트림즈 애플리케이션을 운영하다가 또는 테스트 중에 오프셋을 초기화 해야할 때가 있습니다. 이때는 아래와 같은 명령어를 사용하면 됩니다. $ ./kafka-streams-application-reset.sh --application-id join-application --input-topics streams-test Reset-offsets for input topics [A.s2olap-kakaotalk] Following input topics offsets will be reset to (for consumer group join-application) Topic: streams-test Partition: 2 Offset: 0 Topic: streams-test Partition: 3 Of.. local kafka single broker 띄우기 with 도커 2021. 3. 5. wurstmeister/kafka-docker는 사실상 표준(디팩토)으로 카프카 도커 이미지로 띄우는 용도로 사용하고 있습니다. 사용 방법은 아래와 같습니다. $ git clone https://github.com/wurstmeister/kafka-docker.git $ cd kafka-docker wurstmeister/kafka-docker에 있는 docker-compose-single-broker.yml 의 내부 설정 중 KAFKA_ADVERTISED_HOST_NAME을 변경합니다. $ vi docker-compose-single-broker.yml version: '2' services: zookeeper: image: wurstmeister/zookeeper ports: - "2181:2181".. 특정 시점(날짜+시간)의 레코드부터 가져오도록 설정하기. 2021. 3. 3. 카프카의 토픽에 저장된 레코드는 0.11.0.0 이후부터 timestamp가 저장됩니다. 그러므로 kafka-consumer-group.sh 스크립트로 오프셋을 직접 지정할 수 있는데 --to-datetime을 사용하면 특정 시점(시간) 데이터부터 가져가도록 설정할 수 있습니다. Example) 토픽의 레코드가 다음과 같이 있다고 가정합니다. ConsumerRecord(topic = test, partition = 0, leaderEpoch = 0, offset = 5, CreateTime = 1614748497034, serialized key size = -1, serialized value size = 1, headers = RecordHeaders(headers = [], isReadOnly = .. 아파치 카프카를 데이터 레이크로 사용할 수 있을까? 2021. 2. 25. 데이터 레이크란? 데이터 레이크의 기본 개념은 기업단위의 서비스 데이터들을 한곳의 저장공간에 모아 두는 것입니다. 이렇게 저장된 데이터로부터 BI(Business Intelligence) App으로 조회하거나 시각화, 머신러닝을 수행하였습니다. 데이터 웨어하우스와 다르게 '일단 저장'하고 나서 이후에 스키마를 적용하는 'Schema-on-Read'에 초점을 잡고 있습니다. 데이터 레이크라는 단어가 나온 이래로 지속 발전을 거듭했고 데이터 레이크 기술은 다음과 같이 발전 하였습니다. - 1세대 데이터 레이크 : HDFS, 맵리듀스, 피그, 하이브, 임팔라, 플룸, 스쿱 - 2세대 데이터 레이크 : 클라우드 네이티브로 성장하였고 오브젝트 스토리지(S3), 스파크, 플링크, 프레스토, 스트림셋 스트리밍 데이터.. macOS에서 카프카 버로우 빌드 및 실행하기. 2020. 11. 29. 1. golang 설치 $ brew install go 2. burrow clone from github $ git clone https://github.com/linkedin/Burrow.git 3. go build, install $ cd to the source directory. $ go mod tidy $ go install 3. run $ $GOPATH/bin/Burrow --config-dir=/path/containing/config 버로우를 실행할 때 주의할점은 --config-dir은 말그대로 디렉토리를 설정해야한다. 파일을 설정하면 안된다. burrow.toml이 포함된 디렉토리를 설정한다. 로컬호스트에서 카프카, 주키퍼를 실행할 경우 아래와 같이 burrow.toml을 설정한다. [.. 카프카 토픽의 오프셋 최대 크기는 얼마일까? 2020. 11. 17. 카프카의 토픽에는 파티션이 있습니다. 프로듀서가 레코드를 파티션에 저장하면 각 레코드에는 오프셋이라고 불리는 고유한 번호가 붙게 되는데요. 오프셋 번호가 최대값에 도달하면 어떻게 처리해야하는지 의문을 가질 수 있습니다. 오프셋 번호의 최대값에 대해서 확인해 보았는데요. 오프셋은 int64로 지정되어 있었습니다. private static final Field.Int64 OFFSET = new Field.Int64("offset", "The offset found"); 자바에서 int64의 최대값은 9,223,372,036,854,775,807 입니다. 9,223,372,036,854,775,807는 하루에 1조개의 record를 쌓더라도 25,269년이 걸리는 어마어마한 값입니다. 게다가 파티션을 1개만.. 이전 1 2 3 4 5 6 7 ··· 9 다음 728x90
