Java, Jak uzyskać ilość wiadomości w temacie w apache kafka

To nie jest java, ale może być przydatne

./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}'

Używam tego do porównywania mojego POC. Element, którego chcesz użyć ConsumerOffsetChecker. Możesz go uruchomić za pomocą skryptu bash, jak poniżej.

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

I poniżej jest wynik : Jak widać na czerwonym polu, 999 to numer wiadomości aktualnie w temacie.

Aktualizacja: ConsumerOffsetChecker jest przestarzały od wersji 0.10.0, możesz zacząć używać ConsumerGroupCommand.

Użyj https://prestodb.io/docs/current/connector/kafka-tutorial.html

Silnik super SQL, dostarczany przez Facebook, który łączy się z kilkoma źródłami danych (Cassandra, Kafka, JMX, Redis ...).

PrestoDB działa jako serwer z opcjonalnymi workerami (istnieje tryb autonomiczny bez dodatkowych workerów), następnie do tworzenia zapytań używa się małego pliku wykonywalnego jar (zwanego presto CLI).

Gdy dobrze skonfigurujesz Serwer Presto, możesz użyć tradycyjnego SQL:

SELECT count(*) FROM TOPIC_NAME;

Aby uzyskać wszystkie wiadomości zapisane dla tematu, możesz odszukać konsumenta do początku i końca strumienia dla każdej partycji i zsumować wyniki

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());

Polecenie Apache Kafka, aby uzyskać nie obsługiwane wiadomości na wszystkich partycjach tematu:

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

Druki:

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

Kolumna 6 to wiadomości nieobsługiwane. Dodaj je tak:

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 odczytuje wiersze, pomija linię nagłówka i dodaje szóstą kolumnę, a na końcu wypisuje sumę.

Druki

5

W najnowszych wersjach Kafka Manager znajduje się kolumna zatytułowana podsumowująca Ostatnie przesunięcia .

Czasami warto poznać liczbę wiadomości w każdej partycji, na przykład podczas testowania niestandardowego partycjonera.Kolejne kroki zostały przetestowane pod kątem pracy z Kafka 0.10.2.1-2 z Confluent 3.2. Biorąc pod uwagę temat Kafki, kt i następującą linię poleceń:

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

Który wyświetla Przykładowe wyjście pokazujące liczbę wiadomości w trzech partycjach:

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

Liczba wierszy może być mniej więcej zależna od liczby partycji dla tematu.

Nie próbowałem tego, ale to ma sens.

Możesz również użyć kafka.tools.ConsumerOffsetChecker (źródło).

Używając klienta Javy Kafka 2.11-1.0.0, możesz zrobić następujące rzeczy:

    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));
            }
        }
    }

Wyjście jest takie:

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

./kafka-console-consumer.sh --from-beginning --new-consumer --bootstrap-server yourbroker: 9092 --property print.key = true --property print.value = false --property print.partition --topic yourtopic --timeout-ms 5000 | tail -n 10 / grep "przetworzony w sumie"

Fragmenty Kafka docs

Deprecations in 0.9.0.0

The kafka-consumer-offset-checker.sh kafka.narzędzia.ConsumerOffsetChecker) został wycofany. Idąc do przodu, Proszę użyć kafka-consumer-groups.sh kafka.admin.ConsumerGroupCommand) dla tej funkcjonalności.

Uruchamiam Kafka broker z włączonym SSL zarówno dla serwera, jak i klienta. Poniżej polecenia używam

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

Gdzie /tmp / ssl_config jest jak poniżej

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

Jeśli masz dostęp do interfejsu JMX serwera, przesunięcia początku i końca są obecne w:

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

(musisz wymienić TOPICNAME & PARTITIONNUMBER). Pamiętaj, że musisz sprawdzić dla każdej z replik danej partycji, lub musisz dowiedzieć się, który z brokerów jest liderem dla danej partycji (i to może się zmieniać w czasie).

Alternatywnie, można użyć Kafka Consumer metody beginningOffsets i endOffsets.