apache-kafka

0열

1답변

스프링 카프카 (Spring Kafka) 글로벌 거래 ID는 프로그램이 끝난 후 계속 열어 둔다.

나는 카프카에게 데이터를 보내고 데이터베이스에 쓸 스프링 부트 아래 카프카 스프링 프로듀서를 만들고있다. 나는 그 모든 일을 하나의 거래로 원한다. 나는 카프카에 익숙하지 않고 봄에 대한 전문가도없고 어려움을 겪고 있습니다. 모든 포인터가 많이 감사합니다. 지금까지 나의 코드는 Kafka에 성공적으로 루프를 작성합니다. 나는 아직 DB 을 설정하지 않은,하

3열

2답변

한 카프카 클러스터에서 다른 카프카 클러스터로 메시지 스트리밍

현재 한 카프카 클러스터의 토픽에서 다른 토픽으로 메시지를 쉽게 스트리밍하려고합니다 (원격 -> 로컬 클러스터). 아이디어는 Kafka-Streams를 즉시 사용하여 로컬 클러스터의 실제 메시지를 복제 할 필요가 없지만 Kafka-Streams 처리의 "결과"만 Kafka-Topics에 가져올 수 있습니다. WordCount 데모가 다른 PC의 Kafka-

1열

1답변

사용자 변환을 위해 Kafka 스트림 사용하기

저는 Apache Kafka를 사용하여 ETL 데이터 파이프 라인을 구현했습니다. 나는 Kafka Connect를 추출 및 적재에 사용했습니다. Connect는 원본 데이터를 읽고 카프카 주제의 실제 데이터를 JSON 형식으로 작성합니다. 변환 단계에서 카프카 항목에서 JSON 데이터를 읽고 싶습니다. 그런 다음 일부 맞춤 비즈니스 로직을 기반으로 SQL

0열

1답변

JoinWindows 기간이 스트림 중 하나의 객체에 저장된 2 개의 Kafka KStream을 결합 할 수 있습니까?</li> </ol> <p>사용자 DSL의 API 중 하나 또는 프로세스 API에 그것이 가능 (시간을 시작으로, 종료 시간) (타임 스탬프)</li> <li>번호</p> <ol> <li>시간 창 :

하자 내가 2 개 스트림이 있다고 출력에 TimeWindow에 지정된 시간 범위 내에있는 숫자의 합계를 포함하는 TimeWindow 객체가 포함되도록 스트림을 조인할까요? 구체적으로 말해서 기간 저장소의 위치는 XXX입니다. win.getDuration() 여기서 은이 ValueJoiner에서 참조되는 것입니다. timeWindow.join( nu

0열

1답변

카프카 토폴로지의 소비자 오프셋을 어떻게 변경할 수 있습니까?

요즘 저는 카프카 스트림을 사용하여 메시지를 처리하는 프로젝트를 진행하고 있습니다. 우리의 메시지는 두 개의 식별자로 구성되며, 하나는 사용자의 식별자이고 다른 하나는 사용자가 속한 메시지리스트의 식별자입니다. 응용 프로그램의 이야기는 클라이언트가 내 응용 프로그램을 통해 사용자 목록에 푸시 메시지를 보내려한다는 것입니다. 따라서 클라이언트는 사용자에게 푸

6열

1답변

카프카 다이렉트 스트림이 모든 메시지에 대해 새로운 디코더를 만드는 이유는 무엇입니까?

자바로 작성되고 Spark 2.1을 사용하는 Spark 스트리밍 앱이 있습니다. 카프카에서 메시지를 읽기 위해 KafkaUtils.createDirectStream을 사용하고 있습니다. kafka 메시지에 kryo 인코더/디코더를 사용하고 있습니다. Spark이 메시지를 마이크로 일괄 처리로 가져 오면 메시지가 kryo 디코더를 사용하여 성공적으로 디코딩됩

0열

1답변

하나의 kafka 명령으로 각 주제의 총 메시지 수와 함께 모든 그룹의 모든 주제를 나열하는 방법은 무엇입니까?

각 토픽에 총 메시지 수가있는 모든 그룹 ID의 모든 토픽을 나열하려고하지만 어느 곳에서도 그러한 명령을 찾을 수 없습니다. 아래의 두 명령을 시도했지만 아래 명령의 결과를 단일 명령으로 예상했습니다. 도와주세요. 명령 1 : bin/kafka-topics.sh --list --zookeeper localhost:2181 명령 2 (카프카 서버에있는 모

1열

1답변

응용 프로그램의 상태 저장소 지우기

카프카 스트림 응용 프로그램 내에서 특정 인스턴스의 상태 저장소를 지우고 싶습니다. 예를 들어 상태의 목록 상위 5 개의 최대 값을 유지하는 경우 시간별로 지우고 싶습니다. 그럴 수있는 방법이 있습니까? 응용 프로그램을 중지하지 않고 있습니까?

0열

2답변

카프카 프로듀서를 만들고 send(), flush() 및 close() 메서드를 호출 할 때 올바른 순서는 무엇입니까?

나는 카프카 제작자를 만들고 다른 방법을 호출하는 아래에 언급 된 여러 모델을 가지고 있지만 흐름을 중단해서는 안되며 성능에 영향을 미치지 않도록 프로그램하는 올바른 방법이 무엇인지 확실하지 않습니다. 친절하게 도와주세요. 모델 1 : for(int i=1; i < 100; i++){ Producer<String, String> producer =

1열

1답변

카프카에서 최신 값을 얻으십시오

카프카 주제는 A입니다. 주제 의 데이터 형식은 다음과 같습니다이 { id : 1, name:stackoverflow, created_at:2017-09-28 22:30:00.000} { id : 2, name:confluent, created_at:2017-09-28 22:00:00.000} { id : 3, name:kafka, created_at: