计算 Kafka 主题中的消息答案

【问题标题】：Counting Messages in Kafka Topics计算 Kafka 主题中的消息
【发布时间】：2020-04-21 12:55:34
【问题描述】：

我正在尝试了解如何使用 Kafka 跟踪消息摄取。

我们现在遵循的工作流程是清除主题中的所有消息，然后我们通过代码更改重新摄取。我需要知道这些代码更改有多成功。在当前状态下，我正在使用 Kafka 工具并手动刷新消息总数，并将结果保存在 csv 中，我知道这不是长期可持续的。

您对自动获取 Kafka 主题中的消息计数有何建议？理想情况下，我想以一分钟一分钟的频率点击该主题并获得计数，以及像 1 天这样的时间窗口等。

*由于我们遇到的稳定性问题，我无法使用 KSQL。

【问题讨论】：

标签： apache-kafka kafka-topic

【解决方案1】：

消息计数也取决于日志压缩。

例如，当日志压缩对一个主题有效时，您可能会观察到“奇怪的结果”。假设您有一个主题myTopic，总共有 100 条消息。假设您有一个现在生效的日志压缩策略，该计数可能会减少到 20 条消息，因为旧消息已被压缩。

为了获取每个分区的消息数，可以使用以下命令：

kafka-run-class kafka.tools.GetOffsetShell \
  --broker-list localhost:9092 \
  --topic myTopic

结果将类似于下面的结果（假设myTopic 有 3 个分区）：

myTopic:2:34 
myTopic:1:33 
myTopic:0:33

或者，对于总和，您可以使用这个：

./bin/kafka-run-class.sh kafka.tools.GetOffsetShell \
  --broker-list localhost:2181 \
  --topic myTopic \
  --time -1 \
  --offsets 1 \
  | awk -F  ":" '{sum += $3} END {print sum}'

您可能还会发现一些Kafka Monitoring Tools 在这种情况下很有用。更准确地说，CMAK（又名kafka-manager）在关于Summed Recent Offsets

的指标中有一节

【讨论】：

【解决方案2】：

https://stackoverflow.com/a/63191575/163585

获取kafka中的消息数量：

brokers="<broker1:port>"
topic=<topic-name>
sum_1=$(/usr/hdp/current/kafka-broker/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list $brokers --topic $topic --time -1 | grep -e ':[[:digit:]]*:' | awk -F  ":" '{sum += $3} END {print sum}')
sum_2=$(/usr/hdp/current/kafka-broker/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list $brokers --topic $topic --time -2 | grep -e ':[[:digit:]]*:' | awk -F  ":" '{sum += $3} END {print sum}')
echo "Number of records in topic ${topic}: "$((sum_1 - sum_2))

其中选项--time -1 => 当前最大偏移量&--time -2 是当前最小偏移量。

kafka 中的消息数，在保留期之后，消息将从主题中删除，因此偏移量！= 消息数**

【讨论】：