apache-kafka-streams

0열

1답변

나는 kafka와 관련된 다음과 같은 문제를 해결하려고합니다. 주제가 있습니다. src-topic이라고 부르 자. 이 주제의 기록을 수시로받습니다. 그 값들을 ktable에 저장하고 ktable에 저장된 값을 매 10 초마다 dst-topic에 내보내고 싶습니다. 이 ktable에서 처음으로 값을 내면 나는 내가 보낸 레코드에 1을 더하고 싶습니다. 이후에

0열

1답변

Kafka가 키로 스트림을 합치십시오.

두 개의 토픽이 간단한 키 값 구조 (정수/문자열)를 갖고 있고 완벽하게 작동하는 조인 스트림에서 아주 간단한 예제를했습니다. SELECT * FROM stream1, stream2 WHERE stream1.key = stream2.key AND (stream1.key > 50 && stream1.key < 100) AND (stream2.key > 50

12열

1답변

Kafka Streams API : KStream to KTable

위치 이벤트 (key = user_id, value = user_location)를 보내는 카프카 항목이 있습니다. 나는 KStream로 읽고 그것을 처리 할 수 있어요 : 잘 작동하지만 각 사용자의 마지막으로 알려진 위치와 KTable을 가지고 싶습니다 KStreamBuilder builder = new KStreamBuilder(); KStrea

0열

1답변

스레드 "StreamThread-1"에서 Kafka-Stream 예외 java.lang.IllegalArgumentException : 유효하지 않은 타임 스탬프 -1

기존 주제의 데이터를 다른 주제로 스트리밍하는 간단한 k 스트림 응용 프로그램을 작성 중입니다. : 내가 가진 응용 프로그램을 실행할 때, 그러나 package com.mycompany.app; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.Ka

0열

3답변

Kafka는 키 값이 일반 레코드 인 2 개의 스트림을 조인 할 때 예외를 throw합니다.

키와 값이 GenericRecord 인 두 스트림을 올바르게 조인하기 위해 문제를 해결할 수있는 사람이 있습니까? 먼저, 키와 값에 대해 AVRO 스키마가있는 2 개의 항목을 만듭니다. 이후 나는 스트림을 모두 결합하고있어 출력 항목에 나는 새로운 GenericRecord, 프로젝션 스키마 소위 투사 레코드를 생성하고 코드 후 같이 나는 예외를 가지고 :

0열

1답변

kafka에 작성 kafkastreams를 사용하여 레코드 내용에 기반한 주제

부모의 레코드 내용을 기반으로 카파의 주제 (상위)에서 다른 주제 (하위)로 작성하려고합니다. 부모 항목에서 소비하는 샘플 레코드는 {"date":{"string":"2017-03-20"},"time":{"string":"20:04:13:563"},"event_nr":1572470,"interface":"Transaction Manager","event_i

1열

2답변

실제로 늦게 기록을 삭제하는 방법은 무엇입니까?

여기에 문제가 있습니다 : 1 시간짜리 버킷에서이 숫자 중 MAX를 수집하려고합니다. 여기서 버킷에 최대 3 시간의 지연을 허용합니다. tumbling windows의 경우처럼 들립니다. 는 여기에 지금까지이 작업은 다음과 같습니다 stream.aggregate( () -> 0L, (aggKey, value, aggregate) -> Ma

1열

1답변

KTable에서 키가 누락 된 경우 KStream 핸들 사용

최근 kafka 스트림을 실험하기 시작했습니다. 내가 KStream에 KTable으로 가입해야하는 시나리오가 있습니다. KTable에 일부 키가 들어 있지 않은 경우 일 수 있습니다. 이 경우 나는 NullPointerException이됩니다. 나는 것을 처리 할 수있는 방법을 모르는 java.lang.NullPointerException이 : 특히 내가

4열

1답변

UncaughtExceptionHandler를 사용하여 스트림을 다시 시작하거나 종료하는 올바른 방법

실시간 메시지 변환을위한 드라이버 코드가 아래에있는 스트림 응용 프로그램이 있습니다. String topicName = ... KStreamBuilder builder = new KStreamBuilder(); KStream<String, String> source = builder.stream(topicName); source.transform((

3열

1답변

하나의 마스터 주제에서 여러 스트림

단일 마스터 주제에서 여러 스트림을 만들려면 어떻게해야합니까? KStreamBuilder builder = new KStreamBuilder(); builder.stream(Serdes.String(), Serdes.String(), "master") /* Filtering logic */ .to(Serdes.String(),