将自定义指标与 Amazon Kinesis Data Analytics and Apache Flink 结合使用 - Amazon Kinesis Data Analytics
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅中国的 Amazon Web Services 服务入门

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

将自定义指标与 Amazon Kinesis Data Analytics and Apache Flink 结合使用

适用于 Apache Flink 的 Kinesis Data Analytics 向 CloudWatch 公开了 19 个指标,包括资源使用量和吞吐量指标。此外,您可以创建自己的指标来跟踪应用程序特定的数据,例如处理事件或访问外部资源。

本主题包含下列部分:

工作方式

Kinesis Data Analytics 中的自定义指标使用 Apache Flink 指标系统。Apache Flink 指标具有以下属性:

  • 类型:指标的类型描述了它如何衡量和报告数据。可用的 Apache Flink 指标类型包括计数、仪表、直方图和仪表。有关 Apache Flink 指标类型的更多信息,请参阅指标类型.

    注意

    Amazon CloudWatch 指标不支持直方图 Apache Flink 指标类型。 CloudWatch 只能显示计数、仪表和仪表类型的 Apache Flink 指标。

  • :Scope 指标的范围包括其标识符和一组指示将如何向 CloudWatch 报告指标的键值对。指标的标识符包含以下各项:

    有关指标范围的更多信息,请参阅范围.

有关 Apache Flink 指标的更多信息,请参阅指标中的Apache Flink 文档.

要在适用于 Apache Flink 的 Kinesis Data Analytics 应用程序中创建自定义指标,您可以从任何扩展的用户功能访问 Apache Flink 指标系统RichFunction通过调用GetMetricGroup. 此方法返回MetricGroup您可以用来创建和注册自定义指标的对象。Kinesis Data Analytics 报告使用组密钥创建的所有指标KinesisAnalyticsCloudWatch。您定义的自定义指标具有下列特征:

  • 您的自定义指标有指标名称和组名称。这些名称必须由字母数字字符组成。

  • 您在用户范围中定义的属性(除了KinesisAnalytics指标组) 发布为 CloudWatch 维度。

  • 自定义指标发布在Application默认情况下级别。

  • 根据应用程序的监控级别将维度(任务/操作员/并行度)添加到指标中。您可以使用MonitoringConfiguration的参数CreateApplication操作,或监控配置更新的参数UpdateApplicationaction.

示例

以下代码示例演示了如何创建映射类、创建和增加自定义指标,以及如何通过将映射类添加到DataStream对象。

自定义指标

以下代码示例演示了如何创建映射类来创建计算数据流中记录的指标(功能与numRecordsIn指标):

private static class NoOpMapperFunction extends RichMapFunction<String, String> { private transient int valueToExpose = 0; private final String customMetricName; public NoOpMapperFunction(final String customMetricName) { this.customMetricName = customMetricName; } @Override public void open(Configuration config) { getRuntimeContext().getMetricGroup() .addGroup("kinesisanalytics") .addGroup("Program", "RecordCountApplication") .addGroup("NoOpMapperFunction") .gauge(customMetricName, (Gauge<Integer>) () -> valueToExpose); } @Override public String map(String value) throws Exception { valueToExpose++; return value; } }

在上一示例中,valueToExpose对于应用程序处理的每条记录,变量都会递增。

定义映射类之后,您可以创建一个实现映射的应用程序内流:

DataStream<String> noopMapperFunctionAfterFilter = kinesisProcessed.map(new NoOpMapperFunction("FilteredRecords"));

有关此应用程序的完整代码,请参阅记录计数自定义指标应用.

自定义指标

以下代码示例说明了如何创建映射类,以创建计算数据流中字数的指标:

private static final class Tokenizer extends RichFlatMapFunction<String, Tuple2<String, Integer>> { private transient Counter counter; @Override public void open(Configuration config) { this.counter = getRuntimeContext().getMetricGroup() .addGroup("kinesisanalytics") .addGroup("Service", "WordCountApplication") .addGroup("Tokenizer") .counter("TotalWords"); } @Override public void flatMap(String value, Collector<Tuple2<String, Integer>>out) { // normalize and split the line String[] tokens = value.toLowerCase().split("\\W+"); // emit the pairs for (String token : tokens) { if (token.length() > 0) { counter.inc(); out.collect(new Tuple2<>(token, 1)); } } } }

在上一示例中,counter对于应用程序处理的每个单词,变量都会递增。

定义映射类之后,您可以创建一个实现映射的应用程序内流:

// Split up the lines in pairs (2-tuples) containing: (word,1), and // group by the tuple field "0" and sum up tuple field "1" DataStream<Tuple2<String, Integer>> wordCountStream = input.flatMap(new Tokenizer()).keyBy(0).sum(1); // Serialize the tuple to string format, and publish the output to kinesis sink wordCountStream.map(tuple -> tuple.toString()).addSink(createSinkFromStaticConfig());

有关此应用程序的完整代码,请参阅字数自定义指标应用程序.

查看自定义指标

应用程序的自定义指标显示在 CloudWatch 中的指标控制台AWS/Kinesis分析控制面板,在应用程序指标组。


                控制台屏幕截图,显示创建应用程序页面上的设置。