拓扑统计中奇怪的“发射”数字行为/零统计数字（Storm 1.0.3）

Posted 2023-04-18

【中文标题】拓扑统计中奇怪的“发射”数字行为/零统计数字（Storm 1.0.3）

【英文标题】：Strange 'emitted' numbers behavior / zero stat numbers in Topology stats (Storm 1.0.3)

【发布时间】：2017-04-11 22:41:04

【问题描述】：

This 是我的风暴 UI 统计数据。

问题是我不知道这些数字（发出的元组的）来自哪里。

我的拓扑很简单：kafka spout -> bolt（将数据持久化到 hbase） 拓扑有效 - 当我将数据放入 kafka 主题时，我让它们通过螺栓处理并保存在 hbase 中，然后我使用 hbase shell 中的扫描运算符进行验证（因此正在插入新记录） 但是，每次我将新消息提交到 kafka 并通过螺栓持久化时 - 我的拓扑不会增加“1”发出的数量。 我定期将所有数字增加 20 - 没有向 kafka 发送任何新消息。 IE。我的 kafka 主题几个小时都没有收到任何消息，但是随着时间的推移，发出的元组的数量总是以 20 个为单位增加。我仍然在 hbase 中获得相同数量的记录。 我在 Apache Storm 日志中的任何地方都没有发现异常/错误。 我没有在我的 bolt 实现中执行 ack() 或 fail() 任何元组（这是 BasicBolt 类型自动执行 ack） 即使我在 Kafka 中加载了大量消息，我的 Bolt 指标容量或延迟也始终保持为零 我的 kafka 偏移日志 ($KAFKA/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker) 显示所有消息都已处理，给定主题/组的 Kafka Lag 为 0。

所以我的问题：

那些在 Spout 和 Bolt 中“发射”随着时间增加 20 秒的“隐形”元组是什么？ 是否可以在 Storm UI 中启用“调试”以查看这些元组是什么？ 为什么 Bolt 指标中的容量/延迟始终为零，而 Bolt 被确认为持久化数据？

---

环境详情

我正在使用 Java 8 + Apache Storm 1.0.3

[devops@storm-wk1-prod]~/storm/supervisor/stormdist% storm version
Running: /usr/lib/jvm/jre-1.8.0-openjdk/bin/java -client -Ddaemon.name= -Dstorm.options= -Dstorm.home=/opt/apache-storm-1.0.3 -Dstorm.log.dir=/opt/apache-storm-1.0.3/logs -Djava.library.path=/usr/local/lib:/opt/local/lib:/usr/lib -Dstorm.conf.file= -cp /opt/apache-storm-1.0.3/lib/storm-core-1.0.3.jar:/opt/apache-storm-1.0.3/lib/kryo-3.0.3.jar:/opt/apache-storm-1.0.3/lib/reflectasm-1.10.1.jar:/opt/apache-storm-1.0.3/lib/asm-5.0.3.jar:/opt/apache-storm-1.0.3/lib/minlog-1.3.0.jar:/opt/apache-storm-1.0.3/lib/objenesis-2.1.jar:/opt/apache-storm-1.0.3/lib/clojure-1.7.0.jar:/opt/apache-storm-1.0.3/lib/disruptor-3.3.2.jar:/opt/apache-storm-1.0.3/lib/log4j-api-2.1.jar:/opt/apache-storm-1.0.3/lib/log4j-core-2.1.jar:/opt/apache-storm-1.0.3/lib/log4j-slf4j-impl-2.1.jar:/opt/apache-storm-1.0.3/lib/slf4j-api-1.7.7.jar:/opt/apache-storm-1.0.3/lib/log4j-over-slf4j-1.6.6.jar:/opt/apache-storm-1.0.3/lib/servlet-api-2.5.jar:/opt/apache-storm-1.0.3/lib/storm-rename-hack-1.0.3.jar:/opt/storm/conf org.apache.storm.utils.VersionInfo
Storm 1.0.3
URL https://git-wip-us.apache.org/repos/asf/storm.git -r eac433b0beb3798c4723deb39b3c4fad446378f4
Branch (no branch)
Compiled by ptgoetz on 2017-02-07T20:22Z
From source with checksum c78e52de4b8a22d99551d45dfe9c1a4b

我的storm.yaml：

我正在使用 Storm Supervisor 运行 2 个实例，每个实例具有以下配置：

storm.zookeeper.servers:
  - "10.138.0.8"
  - "10.138.0.9"
  - "10.138.0.16"

storm.zookeeper.port: 2181

nimbus.seeds: ["10.138.0.10"]

storm.local.dir: "/var/log/storm"
supervisor.slots.ports:
  - 6700
  - 6701
  - 6702
  - 6703

worker.childopts: "-Xmx768m"
nimbus.childopts: "-Xmx512m"
supervisor.childopts: "-Xmx256m"

toplogy.yaml

nimbus.host: "10.138.0.10"

# In Storm 0.7.x, this is necessary in order to give workers time to
# initialize. In Storm 0.8.0 and later, it may not be necessary because Storm
# has added a separate, longer timeout for the initial launch of a worker.
supervisor.worker.timeout.secs: 60

topology.workers: 1

拓扑

import tbolts
import tspouts


def create(builder):
    """Create toplogy through Petrel library
    """
    # spout getting data from kafka instance
    # we run 2 tasks of kafka spout
    builder.setSpout("kafka",
                     tspouts.KafkaSpout(), 2)

    # persistence bolt
    # we run 4 tasks of persistence bolt
    builder.setBolt("persistence",
                    tbolts.PersistenceBolt(), 4).shuffleGrouping("kafka")

【参考方案1】：

您的发射计数增加 20 的原因是 Storm 仅在每 20 个元组购买默认值时采样一次以更新其指标。此采样率由topology.stats.sample.rate 配置变量控制，并且可以根据拓扑进行更改。因此，您可以将其设置为1.0（默认为0.05），您将获得准确的发射计数，但这会引入大量的处理开销，并可能导致您的 Acker 和/或指标使用者实例过载.谨慎使用。

【讨论】：

感谢汤姆，但我仍然感到困惑。如果我不发出任何东西（kafka spout 中也没有新消息） - 随着时间的推移，数字仍然会增长 20。所以我确实没有得到处理/排放 - 但它们在离散的时间段内增长 +20 +20。想法？

这可能是由于从各个组件发送的心跳元组。 Storm 拓扑中有隐藏的__system 组件，它们在后台运行，执行事件记录等。这些可能是偶尔出现的幽灵元组的来源，上述采样率将是它跳跃 20 的原因。