2
Ich benutze Kafka Stream und ich versuche, eine KTable in ein Thema zu materialisieren.Apache Kafka Streams Materializing KTables zu einem Thema scheint langsam
Es funktioniert, aber es scheint alle 30 Sekunden oder so getan werden.
Wie/Wann entscheidet sich Kafka Stream, den aktuellen Zustand einer KTable in ein Thema zu materialisieren?
Gibt es eine Möglichkeit, diese Zeit zu verkürzen und sie "in Echtzeit" zu machen?
Hier ist der eigentliche Code I
bin mit// Stream of random ints: (1,1) -> (6,6) -> (3,3)
// one record every 500ms
KStream<Integer, Integer> kStream = builder.stream(Serdes.Integer(), Serdes.Integer(), RandomNumberProducer.TOPIC);
// grouping by key
KGroupedStream<Integer, Integer> byKey = kStream.groupByKey(Serdes.Integer(), Serdes.Integer());
// same behaviour with or without the TimeWindow
KTable<Windowed<Integer>, Long> count = byKey.count(TimeWindows.of(1000L),"total");
// same behaviour with only count.to(Serdes.Integer(), Serdes.Long(), RandomCountConsumer.TOPIC);
count.toStream().map((k,v) -> new KeyValue<>(k.key(), v)).to(Serdes.Integer(), Serdes.Long(), RandomCountConsumer.TOPIC);
Könnten Sie mehr klar sein, auf das, was Sie erreichen wollen? Irgendein Code? Meinst du, du machst so etwas wie: ktable.to ("topic_name")? –
ja das ist genau das, was ich tue –
Ok, ich verstehe dein Problem, aber leider habe ich diesen speziellen Fall noch nie abgestimmt (ich werde tatsächlich etwas Ähnliches brauchen, sobald wir eine Lösung implementieren, die dies erfordert, um sehr zu sein schnell/aktuell) mit dem gesagt, ich würde beginnen mit den verfügbaren Konfigurationen zu spielen: http://kafka.apache.org/documentation/#configuration Möglicherweise zuerst mit dem folgenden: Broker Level: log.flush. * Einsen, Offsets .commit.timeout.ms, Producer Timeouts. Themenebene: Flush. * Einsen Stream-Ebene: commit. * Einsen Wenn Sie eine Lösung finden Sie hier, es wird nützlich sein –