Samza 指標(biāo)

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

指標(biāo)可以以各種方式報告。您可以通過JMX公開它們，這在開發(fā)中很有用。在生產(chǎn)中，常見的設(shè)置是每個Samza容器定期將其指標(biāo)發(fā)布到“指標(biāo)” Kafka 主題，其中來自所有 Samza 作業(yè)的指標(biāo)被聚合。然后，您可以在另一個 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

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

消息看起來像這樣：

{
  "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
    }
  }
}

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

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

您可以通過 MetricsRegistry 注冊自定義指標(biāo)。您的流任務(wù)需要實現(xiàn) InitableTask，以便您可以從 TaskContext 獲取度量注冊表。這個簡單的例子顯示了如何計算任務(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)：計數(shù)器，量表和計時器。當(dāng)您想要了解某些事件的發(fā)生頻率時，請使用計數(shù)器，當(dāng)您想要了解代碼塊花費多少時間時，想要報告某些事件的級別（例如緩沖區(qū)的大小）和計時器的計時器。每個任務(wù)實例（對于每個輸入流分區(qū)）獲取其自己的一組度量。

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

JMX  ?