Java, So erhalten Sie die Anzahl der Nachrichten in einem Thema in Apache Kafka

90

Ich benutze Apache Kafka für Messaging. Ich habe den Produzenten und Konsumenten in Java implementiert. Wie können wir die Anzahl der Nachrichten in einem Thema ermitteln?

Chetan
quelle

Antworten:

31

Der einzige Weg, der aus Verbrauchersicht in den Sinn kommt, besteht darin, die Nachrichten tatsächlich zu konsumieren und sie dann zu zählen.

Der Kafka-Broker stellt JMX-Zähler für die Anzahl der seit dem Start empfangenen Nachrichten bereit, aber Sie können nicht wissen, wie viele davon bereits gelöscht wurden.

In den meisten gängigen Szenarien werden Nachrichten in Kafka am besten als unendlicher Stream angesehen, und es ist nicht relevant, einen diskreten Wert dafür zu erhalten, wie viele Nachrichten derzeit auf der Festplatte gespeichert sind. Darüber hinaus wird es komplizierter, wenn es sich um eine Gruppe von Brokern handelt, die alle eine Teilmenge der Nachrichten in einem Thema enthalten.

Lundahl
quelle
Siehe meine Antwort stackoverflow.com/a/47313863/2017567 . Der Java Kafka-Client ermöglicht das Abrufen dieser Informationen.
Christophe Quintard
93

Es ist kein Java, kann aber nützlich sein

./bin/kafka-run-class.sh kafka.tools.GetOffsetShell 
  --broker-list <broker>:  <port> 
  --topic <topic-name> --time -1 --offsets 1 
  | awk -F  ":" '{sum += $3} END {print sum}'
ssemichev
quelle
9
Sollte dies nicht der Unterschied zwischen frühestem und spätestem Offset pro Partitionssumme sein? bash-4.3# $KAFKA_HOME/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 10.35.25.95:32774 --topic test-topic --time -1 | awk -F ":" '{sum += $3} END {print sum}' 13818663 bash-4.3# $KAFKA_HOME/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 10.35.25.95:32774 --topic test-topic --time -2 | awk -F ":" '{sum += $3} END {print sum}' 12434609 Und dann gibt die Differenz tatsächlich ausstehende Nachrichten im Thema zurück? Hab ich recht?
Kisna
1
Ja das stimmt. Sie müssen eine Differenz berechnen, wenn die frühesten Offsets nicht gleich Null sind.
Ssemichev
Das ist was ich dachte :).
Kisna
1
Gibt es eine Möglichkeit, dies als API und damit in einem Code (JAVA, Scala oder Python) zu verwenden?
Salvob
Hier ist eine Mischung aus meinem Code und Code von Kafka. Es kann nützlich sein. Ich benutzte es für Spark - Streaming - Kafka Integration KafkaClient gist.github.com/ssemichev/c2d94dce7ad65339c9637e1b461f86cf KafkaCluster gist.github.com/ssemichev/fa3605c7b10cb6c7b9c8ab54ffbc5865
ssemichev
15

Ich verwende dies tatsächlich zum Benchmarking meines POC. Das Element, das Sie ConsumerOffsetChecker verwenden möchten. Sie können es mit dem Bash-Skript wie unten ausführen.

bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker  --topic test --zookeeper localhost:2181 --group testgroup

Und unten ist das Ergebnis: Geben Sie hier die Bildbeschreibung ein Wie Sie auf dem roten Feld sehen können, ist 999 die Anzahl der Nachrichten, die sich derzeit im Thema befinden.

Update: ConsumerOffsetChecker ist seit 0.10.0 veraltet. Möglicherweise möchten Sie ConsumerGroupCommand verwenden.

Rudy
quelle
1
Bitte beachten Sie, dass ConsumerOffsetChecker veraltet ist und in Releases nach 0.9.0 gelöscht wird. Verwenden Sie stattdessen ConsumerGroupCommand. (kafka.tools.ConsumerOffsetChecker $)
Szymon Sadło
1
Ja, das habe ich gesagt.
Rudy
Ihr letzter Satz ist nicht korrekt. Der obige Befehl funktioniert immer noch in 0.10.0.1 und die Warnung ist dieselbe wie in meinem vorherigen Kommentar.
Szymon Sadło
13

Manchmal besteht das Interesse darin, die Anzahl der Nachrichten in jeder Partition zu kennen, beispielsweise beim Testen eines benutzerdefinierten Partitionierers. Die folgenden Schritte wurden getestet, um mit Kafka 0.10.2.1-2 aus Confluent 3.2 zu funktionieren. Bei einem Kafka-Thema ktund der folgenden Befehlszeile:

$ kafka-run-class kafka.tools.GetOffsetShell \
  --broker-list host01:9092,host02:9092,host02:9092 --topic kt

Dadurch wird die Beispielausgabe gedruckt, in der die Anzahl der Nachrichten in den drei Partitionen angezeigt wird:

kt:2:6138
kt:1:6123
kt:0:6137

Die Anzahl der Zeilen kann mehr oder weniger abhängig von der Anzahl der Partitionen für das Thema sein.

pdp
quelle
3
Wenn die Protokollkomprimierung aktiviert ist, ergibt das Summieren der Offsets der Partitionen möglicherweise nicht die genaue Anzahl der Nachrichten im Thema.
12

Da dies ConsumerOffsetCheckernicht mehr unterstützt wird, können Sie mit diesem Befehl alle Nachrichten im Thema überprüfen:

bin/kafka-run-class.sh kafka.admin.ConsumerGroupCommand \
    --group my-group \
    --bootstrap-server localhost:9092 \
    --describe

Wo LAGist die Anzahl der Nachrichten in der Themenpartition:

Geben Sie hier die Bildbeschreibung ein

Sie können auch versuchen, Kafkacat zu verwenden . Dies ist ein Open Source-Projekt, mit dem Sie möglicherweise Nachrichten aus einem Thema und einer Partition lesen und in stdout drucken können. Hier ist ein Beispiel, das die letzten 10 Nachrichten aus dem sample-kafka-topicThema liest und dann beendet:

kafkacat -b localhost:9092 -t sample-kafka-topic -p 0 -o -10 -e
Ruslangm
quelle
11

Verwenden Sie https://prestodb.io/docs/current/connector/kafka-tutorial.html

Eine von Facebook bereitgestellte Super-SQL-Engine, die eine Verbindung zu mehreren Datenquellen herstellt (Cassandra, Kafka, JMX, Redis ...).

PrestoDB wird als Server mit optionalen Workern ausgeführt (es gibt einen eigenständigen Modus ohne zusätzliche Worker). Anschließend verwenden Sie eine kleine ausführbare JAR (Presto CLI), um Abfragen durchzuführen.

Sobald Sie den Presto-Server gut konfiguriert haben, können Sie traditionelles SQL verwenden:

SELECT count(*) FROM TOPIC_NAME;
Thomas Decaux
quelle
Dieses Tool ist nett, aber wenn es nicht funktioniert, wenn Ihr Thema mehr als 2 Punkte hat.
Armandfp
7

Apache Kafka-Befehl zum Abrufen nicht behandelter Nachrichten auf allen Partitionen eines Themas:

kafka-run-class kafka.tools.ConsumerOffsetChecker 
    --topic test --zookeeper localhost:2181 
    --group test_group

Drucke:

Group      Topic        Pid Offset          logSize         Lag             Owner
test_group test         0   11051           11053           2               none
test_group test         1   10810           10812           2               none
test_group test         2   11027           11028           1               none

Spalte 6 enthält die nicht behandelten Nachrichten. Addiere sie so:

kafka-run-class kafka.tools.ConsumerOffsetChecker 
    --topic test --zookeeper localhost:2181 
    --group test_group 2>/dev/null | awk 'NR>1 {sum += $6} 
    END {print sum}'

awk liest die Zeilen, überspringt die Kopfzeile und addiert die 6. Spalte und druckt am Ende die Summe.

Druckt

5
Eric Leschinski
quelle
5

Um alle für das Thema gespeicherten Nachrichten abzurufen, können Sie den Konsumenten für jede Partition an den Anfang und das Ende des Streams suchen und die Ergebnisse summieren

List<TopicPartition> partitions = consumer.partitionsFor(topic).stream()
        .map(p -> new TopicPartition(topic, p.partition()))
        .collect(Collectors.toList());
    consumer.assign(partitions); 
    consumer.seekToEnd(Collections.emptySet());
Map<TopicPartition, Long> endPartitions = partitions.stream()
        .collect(Collectors.toMap(Function.identity(), consumer::position));
    consumer.seekToBeginning(Collections.emptySet());
System.out.println(partitions.stream().mapToLong(p -> endPartitions.get(p) - consumer.position(p)).sum());
AutomatedMike
quelle
1
Übrigens, wenn Sie die Komprimierung aktiviert haben, kann es zu Lücken im Stream kommen, sodass die tatsächliche Anzahl der Nachrichten möglicherweise niedriger ist als die hier berechnete Gesamtzahl. Um eine genaue Summe zu erhalten, müssen Sie die Nachrichten erneut abspielen und zählen.
AutomatedMike
5

Führen Sie Folgendes aus (vorausgesetzt, es kafka-console-consumer.shbefindet sich auf dem Pfad):

kafka-console-consumer.sh  --from-beginning \
--bootstrap-server yourbroker:9092 --property print.key=true  \
--property print.value=false --property print.partition \
--topic yourtopic --timeout-ms 5000 | tail -n 10|grep "Processed a total of"
Borislav Markov
quelle
Hinweis: Ich habe die entfernt, --new-consumerda diese Option nicht mehr verfügbar ist (oder anscheinend notwendig ist)
javadba
2

Mit dem Java-Client von Kafka 2.11-1.0.0 können Sie Folgendes tun:

    KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    consumer.subscribe(Collections.singletonList("test"));
    while(true) {
        ConsumerRecords<String, String> records = consumer.poll(100);
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());

            // after each message, query the number of messages of the topic
            Set<TopicPartition> partitions = consumer.assignment();
            Map<TopicPartition, Long> offsets = consumer.endOffsets(partitions);
            for(TopicPartition partition : offsets.keySet()) {
                System.out.printf("partition %s is at %d\n", partition.topic(), offsets.get(partition));
            }
        }
    }

Die Ausgabe ist ungefähr so:

offset = 10, key = null, value = un
partition test is at 13
offset = 11, key = null, value = deux
partition test is at 13
offset = 12, key = null, value = trois
partition test is at 13
Christophe Quintard
quelle
2

In den neuesten Versionen von Kafka Manager gibt es eine Spalte mit dem Titel Summierte letzte Offsets .

Geben Sie hier die Bildbeschreibung ein

f01
quelle
2

Ich hatte dieselbe Frage und so mache ich es von einem KafkaConsumer in Kotlin:

val messageCount = consumer.listTopics().entries.filter { it.key == topicName }
    .map {
        it.value.map { topicInfo -> TopicPartition(topicInfo.topic(), topicInfo.partition()) }
    }.map { consumer.endOffsets(it).values.sum() - consumer.beginningOffsets(it).values.sum()}
    .first()

Sehr grober Code, da ich das gerade zum Laufen gebracht habe, aber im Grunde möchten Sie den Anfangsversatz des Themas vom Endversatz subtrahieren, und dies ist die aktuelle Nachrichtenanzahl für das Thema.

Sie können sich nicht einfach auf den Endversatz verlassen, da andere Konfigurationen (Bereinigungsrichtlinie, Aufbewahrungs-ms usw.) dazu führen können, dass alte Nachrichten aus Ihrem Thema gelöscht werden. Offsets "bewegen" sich nur vorwärts, daher ist es der anfängliche Offset, der sich näher an den Endversatz vorwärts bewegt (oder schließlich an denselben Wert, wenn das Thema gerade keine Nachricht enthält).

Grundsätzlich repräsentiert der Endversatz die Gesamtzahl der Nachrichten, die dieses Thema durchlaufen haben, und die Differenz zwischen den beiden repräsentiert die Anzahl der Nachrichten, die das Thema gerade enthält.

LeYAUable
quelle
0

Auszüge aus Kafka-Dokumenten

Abschreibungen in 0.9.0.0

Die Datei kafka-consumer-offset-checker.sh (kafka.tools.ConsumerOffsetChecker) ist veraltet. Verwenden Sie für diese Funktionalität in Zukunft kafka-consumer-groups.sh (kafka.admin.ConsumerGroupCommand).

Ich verwende Kafka Broker mit aktiviertem SSL für Server und Client. Unter Befehl benutze ich

kafka-consumer-groups.sh --bootstrap-server Broker_IP:Port --list --command-config /tmp/ssl_config kafka-consumer-groups.sh --bootstrap-server Broker_IP:Port --command-config /tmp/ssl_config --describe --group group_name_x

Dabei ist / tmp / ssl_config wie folgt

security.protocol=SSL
ssl.truststore.location=truststore_file_path.jks
ssl.truststore.password=truststore_password
ssl.keystore.location=keystore_file_path.jks
ssl.keystore.password=keystore_password
ssl.key.password=key_password
SR Bandi
quelle
0

Wenn Sie Zugriff auf die JMX-Schnittstelle des Servers haben, sind die Start- und End-Offsets unter folgender Adresse vorhanden:

kafka.log:type=Log,name=LogStartOffset,topic=TOPICNAME,partition=PARTITIONNUMBER
kafka.log:type=Log,name=LogEndOffset,topic=TOPICNAME,partition=PARTITIONNUMBER

(Sie müssen TOPICNAME& ersetzen PARTITIONNUMBER). Denken Sie daran, dass Sie nach den Replikaten einer bestimmten Partition suchen müssen oder herausfinden müssen, welcher der Broker für eine bestimmte Partition führend ist (und dies kann sich im Laufe der Zeit ändern).

Alternativ können Sie Kafka Consumer- Methoden beginningOffsetsund verwenden endOffsets.

Adam Kotwasinski
quelle
-1

Ich habe nicht versucht , diese selbst, aber es scheint Sinn zu machen.

Sie können auch kafka.tools.ConsumerOffsetChecker( Quelle ) verwenden.

hba
quelle
-1

Der einfachste Weg, den ich gefunden habe, besteht darin, die Kafdrop-REST-API zu verwenden /topic/topicNameund den Header key: "Accept"/ value: anzugeben "application/json", um eine JSON-Antwort zurückzugewinnen.

Dies ist hier dokumentiert .

Sebastian Sturza
quelle
-1

Sie können kafkatool verwenden . Bitte überprüfen Sie diesen Link -> http://www.kafkatool.com/download.html

Kafka Tool ist eine GUI-Anwendung zum Verwalten und Verwenden von Apache Kafka-Clustern. Es bietet eine intuitive Benutzeroberfläche, mit der Objekte in einem Kafka-Cluster sowie die in den Themen des Clusters gespeicherten Nachrichten schnell angezeigt werden können.

Vinto
quelle