apache-kafka-streams

2열

1답변

Kafka는 결합 된 로그 이벤트에 단일 로그 이벤트 라인을 집계합니다.

카프카를 사용하여 로그 이벤트를 처리하고 있습니다. 나는 간단한 커넥터와 스트림 변환을위한 Kafka Connect와 Kafka Streams에 대한 기본적인 지식을 가지고 있습니다. 이제 I는 다음과 같은 구조의 로그 파일을 가지고 timestamp event_id event 로그의 이벤트 EVENT_ID로 연결된 복수의 기록 라인 (예를 들면 메일

4열

1답변

스트림 처리의 비 결정적 함수

StreamScope와 같은 일부 시스템에서는 함수가 결정적이어야하며 (처리되는 순서와 마찬가지로) 함수가 스트림에있을 때 각 메시지의 시퀀스 번호가 있기 때문입니다. 실패의 경우,이 시퀀스 번호는 이벤트가 다시 계산되어야하는지 (스트림에 의해 지속 되었기 때문에) 재전송 노드가 동일한 이벤트를 두 번 계산하지 않도록 결정하는 데 사용됩니다. Flink,

3열

1답변

KStream (Confluent Platform)에서 오프셋 값을 얻는 방법

나는 Kafka Streams로 PoC를 개발 중입니다. 이제 스트림 소비자에서 오프셋 값을 가져와 각 메시지에 대해 고유 한 키 (topic-offset)->hash을 생성해야합니다. 그 이유는 생산자가 syslog이고 ID 중 일부만이 ID를 가지고 있기 때문입니다. 재 처리의 경우 동일한 키를 재생성해야하기 때문에 소비자에 UUID를 생성 할 수 없습

3열

1답변

Kafka 스트림을 사용하는 프로세서로 키와 값을 필터링하는 방법 DSL

StateStore와 상호 작용하여 메시지를 필터링하고 복잡한 논리를 처리하는 프로세서가 있습니다. process(key,value) 메서드에서는 context.forward(key,value)을 사용하여 필요한 키와 값을 보냅니다. 디버깅 목적으로도 인쇄합니다. 다른 두 스트림을 결합한 결과 KStream mergedStream이 있습니다. 그 스트림의

8열

2답변

카프카 스트리밍 동시성?

하나의 주제에서 레코드를 읽고 처리하고 다른 주제로 레코드를 출력하는 기본 Kafka 스트리밍 코드가 있습니다. 카프카 스트리밍이 동시성을 어떻게 처리합니까? 모든 것이 단일 스레드에서 실행됩니까? 설명서에 언급 된 내용이 보이지 않습니다. 단일 스레드 인 경우 많은 양의 데이터를 처리하기 위해 다중 스레드 처리 옵션을 원합니다. 멀티 스레드 인 경우 SQ

2열

1답변

카프카 스트림 - kafka-streams-application-reset.sh

카프카 0.10.0.1이 kafka-streams-application-reset.sh 플루라는 스크립트를 사용하여 카프카 스트림 응용 프로그램을 재설정 할 수있는 기능을 추가 잘못된 API 버전을 보내는이 스크립트 here에 대한 몇 가지 좋은 문서를 가지고 . 불행하게도, 로컬 스크립트를 실행하면 오류가 발생합니다 $ bin/kafka-streams-a

2열

1답변

카프카 - 필터와 필 터폰을 동시에 사용하는 방법은 무엇입니까?

주제에서 데이터를 가져와 두 가지 주제로 필터링해야하는 Kafka 스트림이 있습니다. 나는 이런 식으로 작업을 수행 할 때 KStream<String, Model> stream = builder.stream(Serdes.String(), specificAvroSerde, "not-filtered-topic"); stream.filter((key, valu

4열

1답변

Apache Kafka Streams가 RocksDB를 사용하는 이유와 어떻게 그것을 변경할 수 있습니까?

Apache Kafka 0.9 및 0.10의 새로운 기능을 조사하는 동안 우리는 KStreams 및 KTables를 사용했습니다. Kafka가 RocksDB를 내부적으로 사용한다는 흥미로운 사실은 입니다. Introducing Kafka Streams: Stream Processing Made Simple을 참조하십시오. RocksDB는 JVN 호환 언어로

1열

1답변

Kafka Streams DSL 프로세서에 사용자 정의 StateStore를 추가하는 방법은 무엇입니까?

내 Kafka 스트림 앱 중 하나 인 경우 DSL과 프로세서 API의 기능을 모두 사용해야합니다. 내 스트리밍 앱 흐름은 source -> selectKey -> filter -> aggregate (on a window) -> sink 입니다. 집계 후 단일 집계 메시지를 싱크대로 보내야합니다. 그래서 KStreamBuilder builder = ne

5열

3답변

카프카 스트리밍 + 스파크 스트리밍 + 기계 학습

데이터 스트림에서 기계 학습 알고리즘을 실행하기 위해 POC를하고 있습니다. 초기 생각은 데이터를 가져 와서 Spark Streaming -> 여러 테이블의 데이터 집계 -> 데이터 스트림에서 MLLib 실행 -> 출력 생성을 사용하는 것이 었습니다. 하지만 나는 Confluent 's Kafka Streaming을 가로 질러 캠을 질렀습니다. 이제 혼란스