Samza 指标
当您在生产过程中运行流程时，您必须确保良好的指标来跟踪您的工作状况。为了使这个更容易，Samza 包括一个指标库。Samza 本身使用它来生成一些标准指标，例如消息吞吐量，但是您也可以在任务代码中使用它来发出自定义指标。
指标可以以各种方式报告。您可以通过JMX公开它们，这在开发中很有用。在生产中，常见的设置是每个Samza容器定期将其指标发布到“指标” Kafka 主题，其中来自所有 Samza 作业的指标被聚合。然后，您可以在另一个 Samza 作业中使用此流，并将指标发送到您最喜欢的图形系统，如 Graphite。
要设置工作以将指标发布到 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通过此配置，作业会自动将几个 JSON 编码的消息发送到 Kafka 的每个60秒的“指标”主题。
消息看起来像这样：
{
"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
}
}
}每个任务实例都有一个单独的消息，标题会告诉您任务的作业名称，作业 ID 和分区。这些指标可让您查看已处理和发送的消息数量，输入流分区中的当前偏移量以及其他详细信息。还有其他消息可以提供关于 JVM（堆大小，垃圾收集信息，线程等），Kafka 生产者和消费者的内部度量等指标。通过 samza 发出的所有指标的列表显示在这里。
如果您有一些值得您关注的话，可以轻松地在您的工作中生成自定义指标。您可以使用 Samza 的内置指标框架，其设计类似于 Coda Hale 的指标库。
您可以通过 MetricsRegistry 注册自定义指标。您的流任务需要实现 InitableTask，以便您可以从 TaskContext 获取度量注册表。这个简单的例子显示了如何计算任务处理的消息数：
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 目前支持三种指标：计数器，量表和计时器。当您想要了解某些事件的发生频率时，请使用计数器，当您想要了解代码块花费多少时间时，想要报告某些事件的级别（例如缓冲区的大小）和计时器的计时器。每个任务实例（对于每个输入流分区）获取其自己的一组度量。
如果要以其他方式报告指标，例如直接进入图形系统（不需要通过 Kafka），则可以实现 MetricsReporterFactory 并在作业配置中引用它。
