Samza 指標(biāo)

當(dāng)您在生產(chǎn)過(guò)程中運(yùn)行流程時(shí)，您必須確保良好的指標(biāo)來(lái)跟蹤您的工作狀況。為了使這個(gè)更容易，Samza 包括一個(gè)指標(biāo)庫(kù)。Samza 本身使用它來(lái)生成一些標(biāo)準(zhǔn)指標(biāo)，例如消息吞吐量，但是您也可以在任務(wù)代碼中使用它來(lái)發(fā)出自定義指標(biāo)。

指標(biāo)可以以各種方式報(bào)告。您可以通過(guò)JMX公開(kāi)它們，這在開(kāi)發(fā)中很有用。在生產(chǎn)中，常見(jiàn)的設(shè)置是每個(gè)Samza容器定期將其指標(biāo)發(fā)布到“指標(biāo)” Kafka 主題，其中來(lái)自所有 Samza 作業(yè)的指標(biāo)被聚合。然后，您可以在另一個(gè) Samza 作業(yè)中使用此流，并將指標(biāo)發(fā)送到您最喜歡的圖形系統(tǒng)，如 Graphite。

要設(shè)置工作以將指標(biāo)發(fā)布到 Kafka，可以使用以下配置：

# Define a metrics reporter called "snapshot", which publishes metrics
# every 60 seconds.
metrics.reporters=snapshot
metrics.reporter.snapshot.class=org.apache.samza.metrics.reporter.MetricsSnapshotReporterFactory

# Tell the snapshot reporter to publish to a topic called "metrics"
# in the "kafka" system.
metrics.reporter.snapshot.stream=kafka.metrics

# Encode metrics data as JSON.
serializers.registry.metrics.class=org.apache.samza.serializers.MetricsSnapshotSerdeFactory
systems.kafka.streams.metrics.samza.msg.serde=metrics

通過(guò)此配置，作業(yè)會(huì)自動(dòng)將幾個(gè) JSON 編碼的消息發(fā)送到 Kafka 的每個(gè)60秒的“指標(biāo)”主題。

消息看起來(lái)像這樣：

{
  "header": {
    "container-name": "samza-container-0",
    "host": "samza-grid-1234.example.com",
    "job-id": "1",
    "job-name": "my-samza-job",
    "reset-time": 1401729000347,
    "samza-version": "0.0.1",
    "source": "Partition-2",
    "time": 1401729420566,
    "version": "0.0.1"
  },
  "metrics": {
    "org.apache.samza.container.TaskInstanceMetrics": {
      "commit-calls": 7,
      "commit-skipped": 77948,
      "kafka-input-topic-offset": "1606",
      "messages-sent": 985,
      "process-calls": 1093,
      "send-calls": 985,
      "send-skipped": 76970,
      "window-calls": 0,
      "window-skipped": 77955
    }
  }
}

每個(gè)任務(wù)實(shí)例都有一個(gè)單獨(dú)的消息，標(biāo)題會(huì)告訴您任務(wù)的作業(yè)名稱(chēng)，作業(yè) ID 和分區(qū)。這些指標(biāo)可讓您查看已處理和發(fā)送的消息數(shù)量，輸入流分區(qū)中的當(dāng)前偏移量以及其他詳細(xì)信息。還有其他消息可以提供關(guān)于 JVM（堆大小，垃圾收集信息，線程等），Kafka 生產(chǎn)者和消費(fèi)者的內(nèi)部度量等指標(biāo)。通過(guò) samza 發(fā)出的所有指標(biāo)的列表顯示在這里。

如果您有一些值得您關(guān)注的話(huà)，可以輕松地在您的工作中生成自定義指標(biāo)。您可以使用 Samza 的內(nèi)置指標(biāo)框架，其設(shè)計(jì)類(lèi)似于 Coda Hale 的指標(biāo)庫(kù)。

您可以通過(guò) MetricsRegistry 注冊(cè)自定義指標(biāo)。您的流任務(wù)需要實(shí)現(xiàn) InitableTask，以便您可以從 TaskContext 獲取度量注冊(cè)表。這個(gè)簡(jiǎn)單的例子顯示了如何計(jì)算任務(wù)處理的消息數(shù)：

public class MyJavaStreamTask implements StreamTask, InitableTask {
  private Counter messageCount;

  public void init(Config config, TaskContext context) {
    this.messageCount = context
      .getMetricsRegistry()
      .newCounter(getClass().getName(), "message-count");
  }

  public void process(IncomingMessageEnvelope envelope,
                      MessageCollector collector,
                      TaskCoordinator coordinator) {
    messageCount.inc();
  }
}

Samza 目前支持三種指標(biāo)：計(jì)數(shù)器，量表和計(jì)時(shí)器。當(dāng)您想要了解某些事件的發(fā)生頻率時(shí)，請(qǐng)使用計(jì)數(shù)器，當(dāng)您想要了解代碼塊花費(fèi)多少時(shí)間時(shí)，想要報(bào)告某些事件的級(jí)別（例如緩沖區(qū)的大小）和計(jì)時(shí)器的計(jì)時(shí)器。每個(gè)任務(wù)實(shí)例（對(duì)于每個(gè)輸入流分區(qū)）獲取其自己的一組度量。

如果要以其他方式報(bào)告指標(biāo)，例如直接進(jìn)入圖形系統(tǒng)（不需要通過(guò) Kafka），則可以實(shí)現(xiàn) MetricsReporterFactory 并在作業(yè)配置中引用它。

JMX  ?