# Metrics and dimensions in Managed Service for Apache Flink When your Managed Service for Apache Flink processes a data source, Managed Service for Apache Flink reports the following metrics and dimensions to Amazon CloudWatch. **Flink 2.2 metric changes** Flink 2.2 introduces metric changes that may affect your monitoring and alarms. Review the following changes before upgrading: The `fullRestarts` metric has been removed. Use `numRestarts` instead. The `uptime` and `downtime` metrics are deprecated and will be removed in a future release. Migrate to the new state-specific metrics. The `bytesRequestedPerFetch` metric for Kinesis Data Streams connector 6.0.0 has been removed. ## Application metrics | Metric | Unit | Description | Level | Usage Notes | | --- | --- | --- | --- | --- | | backPressuredTimeMsPerSecond\$1 | Milliseconds | The time (in milliseconds) this task or operator is back pressured per second. | Task, Operator, Parallelism | \$1Available for Managed Service for Apache Flink applications running Flink version 1.13 only. These metrics can be useful in identifying bottlenecks in an application. | | busyTimeMsPerSecond\$1 | Milliseconds | The time (in milliseconds) this task or operator is busy (neither idle nor back pressured) per second. Can be NaN, if the value could not be calculated. | Task, Operator, Parallelism | \$1Available for Managed Service for Apache Flink applications running Flink version 1.13 only. These metrics can be useful in identifying bottlenecks in an application. | | cpuUtilization | Percentage | Overall percentage of CPU utilization across task managers. For example, if there are five task managers, Managed Service for Apache Flink publishes five samples of this metric per reporting interval. | Application | You can use this metric to monitor minimum, average, and maximum CPU utilization in your application. The CPUUtilization metric only accounts for CPU usage of the TaskManager JVM process running inside the container. | | containerCPUUtilization | Percentage | Overall percentage of CPU utilization across task manager containers in Flink application cluster. For example, if there are five task managers, correspondingly there are five TaskManager containers and Managed Service for Apache Flink publishes 2 \$1 five samples of this metric per 1 minute reporting interval. | Application | It is calculated per container as: *Total CPU time (in seconds) consumed by container \$1 100 / Container CPU limit (in CPUs/seconds)* The `CPUUtilization` metric only accounts for CPU usage of the TaskManager JVM process running inside the container. There are other components running outside the JVM within the same container. The `containerCPUUtilization` metric gives you a more complete picture, including all processes in terms of CPU exhaustion at the container and failures resulting from that. | | containerMemoryUtilization | Percentage | Overall percentage of memory utilization across task manager containers in Flink application cluster. For example, if there are five task managers, correspondingly there are five TaskManager containers and Managed Service for Apache Flink publishes 2 \$1 five samples of this metric per 1 minute reporting interval. | Application | It is calculated per container as: *Container memory usage (bytes) \$1 100 / Container memory limit as per pod deployment spec (in bytes)* The `HeapMemoryUtilization` and `ManagedMemoryUtilzations` metrics only account for specific memory metrics like Heap Memory Usage of TaskManager JVM or Managed Memory (memory usage outside JVM for native processes like [RocksDB State Backend](https://flink.apache.org/2021/01/18/rocksdb.html#:~:text=Conclusion-,The%20RocksDB%20state%20backend%20(i.e.%2C%20RocksDBStateBackend)%20is%20one%20of,with%20exactly%2Donce%20processing%20guarantees.)). The `containerMemoryUtilization` metric gives you a more complete picture by including the working set memory, which is a better tracker of total memory exhaustion. Upon its exhaustion, it will result in `Out of Memory Error` for the TaskManager pod. | | containerDiskUtilization | Percentage | Overall percentage of disk utilization across task manager containers in Flink application cluster. For example, if there are five task managers, correspondingly there are five TaskManager containers and Managed Service for Apache Flink publishes 2 \$1 five samples of this metric per 1 minute reporting interval. | Application | It is calculated per container as: *Disk usage in bytes \$1 100 / Disk Limit for container in bytes* For containers, it represents utilization of the filesystem on which root volume of the container is set up. | | currentInputWatermark | Milliseconds | The last watermark this application/operator/task/thread has received | Application, Operator, Task, Parallelism | This record is only emitted for dimensions with two inputs. This is the minimum value of the last received watermarks. | | currentOutputWatermark | Milliseconds | The last watermark this application/operator/task/thread has emitted | Application, Operator, Task, Parallelism | | | downtime [DEPRECATED] | Milliseconds | For jobs currently in a failing/recovering situation, the time elapsed during this outage. | Application | This metric measures the time elapsed while a job is failing or recovering. This metric returns 0 for running jobs and -1 for completed jobs. If this metric is not 0 or -1, this indicates that the Apache Flink job for the application failed to run. **Deprecated in Flink 2.2.** Use `restartingTime`, `cancellingTime`, and/or `failingTime` instead. | | failingTime | Milliseconds | The time (in milliseconds) that the application has spent in a failing state. Use this metric to monitor application failures and trigger alerts. | Application, Flow | Available from Flink 2.2. Replaces part of the deprecated downtime metric. | | heapMemoryUtilization | Percentage | Overall heap memory utilization across task managers. For example, if there are five task managers, Managed Service for Apache Flink publishes five samples of this metric per reporting interval. | Application | You can use this metric to monitor minimum, average, and maximum heap memory utilization in your application. The HeapMemoryUtilization only accounts for specific memory metrics like Heap Memory Usage of TaskManager JVM. | | idleTimeMsPerSecond\$1 | Milliseconds | The time (in milliseconds) this task or operator is idle (has no data to process) per second. Idle time excludes back pressured time, so if the task is back pressured it is not idle. | Task, Operator, Parallelism | \$1Available for Managed Service for Apache Flink applications running Flink version 1.13 only. These metrics can be useful in identifying bottlenecks in an application. | | lastCheckpointSize | Bytes | The total size of the last checkpoint | Application | You can use this metric to determine running application storage utilization. If this metric is increasing in value, this may indicate that there is an issue with your application, such as a memory leak or bottleneck. | | lastCheckpointDuration | Milliseconds | The time it took to complete the last checkpoint | Application | This metric measures the time it took to complete the most recent checkpoint. If this metric is increasing in value, this may indicate that there is an issue with your application, such as a memory leak or bottleneck. In some cases, you can troubleshoot this issue by disabling checkpointing. | | managedMemoryUsed\$1 | Bytes | The amount of managed memory currently used. | Application, Operator, Task, Parallelism | \$1Available for Managed Service for Apache Flink applications running Flink version 1.13 only. This relates to memory managed by Flink outside the Java heap. It is used for the RocksDB state backend, and is also available to applications. | | managedMemoryTotal\$1 | Bytes | The total amount of managed memory. | Application, Operator, Task, Parallelism | \$1Available for Managed Service for Apache Flink applications running Flink version 1.13 only. This relates to memory managed by Flink outside the Java heap. It is used for the RocksDB state backend, and is also available to applications. The `ManagedMemoryUtilzations` metric only accounts for specific memory metrics like Managed Memory (memory usage outside JVM for native processes like [RocksDB State Backend](https://flink.apache.org/2021/01/18/rocksdb.html#:~:text=Conclusion-,The%20RocksDB%20state%20backend%20(i.e.%2C%20RocksDBStateBackend)%20is%20one%20of,with%20exactly%2Donce%20processing%20guarantees.)) | | managedMemoryUtilization\$1 | Percentage | Derived by managedMemoryUsed/managedMemoryTotal | Application, Operator, Task, Parallelism | \$1Available for Managed Service for Apache Flink applications running Flink version 1.13 only. This relates to memory managed by Flink outside the Java heap. It is used for the RocksDB state backend, and is also available to applications. | | numberOfFailedCheckpoints | Count | The number of times checkpointing has failed. | Application | You can use this metric to monitor application health and progress. Checkpoints may fail due to application problems, such as throughput or permissions issues. | | numRecordsIn\$1 | Count | The total number of records this application, operator, or task has received. | Application, Operator, Task, Parallelism | \$1To apply the SUM statistic over a period of time (second/minute): [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/managed-flink/latest/java/metrics-dimensions.html) The metric's Level specifies whether this metric measures the total number of records the entire application, a specific operator, or a specific task has received. | | numRecordsInPerSecond\$1 | Count/Second | The total number of records this application, operator or task has received per second. | Application, Operator, Task, Parallelism | \$1To apply the SUM statistic over a period of time (second/minute): [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/managed-flink/latest/java/metrics-dimensions.html) The metric's Level specifies whether this metric measures the total number of records the entire application, a specific operator, or a specific task has received per second. | | numRecordsOut\$1 | Count | The total number of records this application, operator or task has emitted. | Application, Operator, Task, Parallelism | \$1To apply the SUM statistic over a period of time (second/minute): [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/managed-flink/latest/java/metrics-dimensions.html) The metric's Level specifies whether this metric measures the total number of records the entire application, a specific operator, or a specific task has emitted. | | numLateRecordsDropped\$1 | Count | Application, Operator, Task, Parallelism | | \$1To apply the SUM statistic over a period of time (second/minute): [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/managed-flink/latest/java/metrics-dimensions.html) The number of records this operator or task has dropped due to arriving late. | | numRecordsOutPerSecond\$1 | Count/Second | The total number of records this application, operator or task has emitted per second. | Application, Operator, Task, Parallelism | \$1To apply the SUM statistic over a period of time (second/minute): [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/managed-flink/latest/java/metrics-dimensions.html) The metric's Level specifies whether this metric measures the total number of records the entire application, a specific operator, or a specific task has emitted per second. | | oldGenerationGCCount | Count | The total number of old garbage collection operations that have occurred across all task managers. | Application | | | oldGenerationGCTime | Milliseconds | The total time spent performing old garbage collection operations. | Application | You can use this metric to monitor sum, average, and maximum garbage collection time. | | threadsCount | Count | The total number of live threads used by the application. | Application | This metric measures the number of threads used by the application code. This is not the same as application parallelism. | | cancellingTime | Milliseconds | The time (in milliseconds) that the application has spent in a cancelling state. Use this metric to monitor application cancellation operations. | Application, Flow | Available from Flink 2.2. Replaces part of the deprecated downtime metric. | | restartingTime | Milliseconds | The time (in milliseconds) that the application has spent in a restarting state. Use this metric to monitor application restart behavior. | Application, Flow | Available from Flink 2.2. Replaces part of the deprecated downtime metric. | | runningTime | Milliseconds | The time (in milliseconds) that the application has been running without interruption. Replaces the deprecated uptime metric. | Application, Flow | Available from Flink 2.2. Use as a direct replacement for the deprecated uptime metric. | | uptime [DEPRECATED] | Milliseconds | The time that the job has been running without interruption. | Application | You can use this metric to determine if a job is running successfully. This metric returns -1 for completed jobs. **Deprecated in Flink 2.2.** Use `runningTime` instead. | | jobmanagerFileDescriptorsMax | Count | The maximum number of file descriptors available to the JobManager. | Application, Flow, Host | Use this metric to monitor file descriptor capacity. | | jobmanagerFileDescriptorsOpen | Count | The current number of open file descriptors for the JobManager. | Application, Flow, Host | Use this metric to monitor file descriptor usage and detect potential resource exhaustion. | | taskmanagerFileDescriptorsMax | Count | The maximum number of file descriptors available to each TaskManager. | Application, Flow, Host, tm\$1id | Use this metric to monitor file descriptor capacity. | | taskmanagerFileDescriptorsOpen | Count | The current number of open file descriptors for each TaskManager. | Application, Flow, Host, tm\$1id | Use this metric to monitor file descriptor usage and detect potential resource exhaustion. | | KPUs\$1 | Count | The total number of KPUs used by the application. | Application | \$1This metric receives one sample per billing period (one hour). To visualize the number of KPUs over time, use MAX or AVG over a period of at least one (1) hour. The KPU count includes the `orchestration` KPU. For more information, see [Managed Service for Apache Flink Pricing](https://aws.amazon.com/managed-service-apache-flink/pricing/). | **Flink 2.2 metric migration guidance** **Migration from fullRestarts:** The `fullRestarts` metric has been removed in Flink 2.2. Use the `numRestarts` metric instead. The `numRestarts` metric provides equivalent functionality and can be used as a direct replacement in CloudWatch alarms without requiring threshold adjustments. **Migration from uptime:** The `uptime` metric is deprecated in Flink 2.2 and will be removed in a future release. Use the `runningTime` metric instead. The `runningTime` metric provides equivalent functionality and can be used as a direct replacement in CloudWatch alarms without requiring threshold adjustments. **Migration from downtime:** The `downtime` metric is deprecated in Flink 2.2 and will be removed in a future release. Depending on what you want to monitor, use one or more of the following metrics: `restartingTime`: Monitor time spent restarting the application `cancellingTime`: Monitor time spent cancelling the application `failingTime`: Monitor time spent in a failing state ## Kinesis Data Streams connector metrics AWS emits all records for Kinesis Data Streams in addition to the following: | Metric | Unit | Description | Level | Usage Notes | | --- | --- | --- | --- | --- | | millisbehindLatest | Milliseconds | The number of milliseconds the consumer is behind the head of the stream, indicating how far behind current time the consumer is. | Application (for Stream), Parallelism (for ShardId) | [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/managed-flink/latest/java/metrics-dimensions.html) | **Note** The `bytesRequestedPerFetch` metric has been removed in Flink AWS connector version 6.0.0 (the only connector version compatible with Flink 2.2). The only Kinesis Data Streams connector metric available in Flink 2.2 is `millisBehindLatest`. ## Amazon MSK connector metrics AWS emits all records for Amazon MSK in addition to the following: | Metric | Unit | Description | Level | Usage Notes | | --- | --- | --- | --- | --- | | currentoffsets | N/A | The consumer's current read offset, for each partition. A particular partition's metric can be specified by topic name and partition id. | Application (for Topic), Parallelism (for PartitionId) | | | commitsFailed | N/A | The total number of offset commit failures to Kafka, if offset committing and checkpointing are enabled. | Application, Operator, Task, Parallelism | Committing offsets back to Kafka is only a means to expose consumer progress, so a commit failure does not affect the integrity of Flink's checkpointed partition offsets. | | commitsSucceeded | N/A | The total number of successful offset commits to Kafka, if offset committing and checkpointing are enabled. | Application, Operator, Task, Parallelism | | | committedoffsets | N/A | The last successfully committed offsets to Kafka, for each partition. A particular partition's metric can be specified by topic name and partition id. | Application (for Topic), Parallelism (for PartitionId) | | | records\$1lag\$1max | Count | The maximum lag in terms of number of records for any partition in this window | Application, Operator, Task, Parallelism | | | bytes\$1consumed\$1rate | Bytes | The average number of bytes consumed per second for a topic | Application, Operator, Task, Parallelism | | ## Apache Zeppelin metrics For Studio notebooks, AWS emits the following metrics at the application level: `KPUs`, `cpuUtilization`, `heapMemoryUtilization`, `oldGenerationGCTime`, `oldGenerationGCCount`, and `threadCount`. In addition, it emits the metrics shown in the following table, also at the application level. **** | Metric | Unit | Description | Prometheus name | | --- | --- | --- | --- | | zeppelinCpuUtilization | Percentage | Overall percentage of CPU utilization in the Apache Zeppelin server. | process\$1cpu\$1usage | | zeppelinHeapMemoryUtilization | Percentage | Overall percentage of heap memory utilization for the Apache Zeppelin server. | jvm\$1memory\$1used\$1bytes | | zeppelinThreadCount | Count | The total number of live threads used by the Apache Zeppelin server. | jvm\$1threads\$1live\$1threads | | zeppelinWaitingJobs | Count | The number of queued Apache Zeppelin jobs waiting for a thread. | jetty\$1threads\$1jobs | | zeppelinServerUptime | Seconds | The total time that the server has been up and running. | process\$1uptime\$1seconds | # View CloudWatch metrics You can view CloudWatch metrics for your application using the Amazon CloudWatch console or the AWS CLI. **To view metrics using the CloudWatch console** 1. Open the CloudWatch console at [https://console.aws.amazon.com/cloudwatch/](https://console.aws.amazon.com/cloudwatch/). 1. In the navigation pane, choose **Metrics**. 1. In the **CloudWatch Metrics by Category** pane for Managed Service for Apache Flink, choose a metrics category. 1. In the upper pane, scroll to view the full list of metrics. **To view metrics using the AWS CLI** + At a command prompt, use the following command. ``` 1. aws cloudwatch list-metrics --namespace "AWS/KinesisAnalytics" --region region ``` # Set CloudWatch metrics reporting levels You can control the level of application metrics that your application creates. Managed Service for Apache Flink supports the following metrics levels: + **Application:** The application only reports the highest level of metrics for each application. Managed Service for Apache Flink metrics are published at the Application level by default. + **Task:** The application reports task-specific metric dimensions for metrics defined with the Task metric reporting level, such as number of records in and out of the application per second. + **Operator:** The application reports operator-specific metric dimensions for metrics defined with the Operator metric reporting level, such as metrics for each filter or map operation. + **Parallelism:** The application reports `Task` and `Operator` level metrics for each execution thread. This reporting level is not recommended for applications with a Parallelism setting above 64 due to excessive costs. **Note** You should only use this metric level for troubleshooting because of the amount of metric data that the service generates. You can only set this metric level using the CLI. This metric level is not available in the console. The default level is **Application**. The application reports metrics at the current level and all higher levels. For example, if the reporting level is set to **Operator**, the application reports **Application**, **Task**, and **Operator** metrics. You set the CloudWatch metrics reporting level using the `MonitoringConfiguration` parameter of the [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_CreateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_CreateApplication.html) action, or the `MonitoringConfigurationUpdate` parameter of the [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html) action. The following example request for the [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html) action sets the CloudWatch metrics reporting level to **Task**: ``` { "ApplicationName": "MyApplication", "CurrentApplicationVersionId": 4, "ApplicationConfigurationUpdate": { "FlinkApplicationConfigurationUpdate": { "MonitoringConfigurationUpdate": { "ConfigurationTypeUpdate": "CUSTOM", "MetricsLevelUpdate": "TASK" } } } } ``` You can also configure the logging level using the `LogLevel` parameter of the [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_CreateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_CreateApplication.html) action or the `LogLevelUpdate` parameter of the [https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html](https://docs.aws.amazon.com/managed-service-for-apache-flink/latest/apiv2/API_UpdateApplication.html) action. You can use the following log levels: + `ERROR`: Logs potentially recoverable error events. + `WARN`: Logs warning events that might lead to an error. + `INFO`: Logs informational events. + `DEBUG`: Logs general debugging events. For more information about Log4j logging levels, see [Custom Log Levels](https://logging.apache.org/log4j/2.x/manual/customloglevels.html) in the [Apache Log4j](https://logging.apache.org/log4j/2.x/) documentation. # Use custom metrics with Amazon Managed Service for Apache Flink Managed Service for Apache Flink exposes 19 metrics to CloudWatch, including metrics for resource usage and throughput. In addition, you can create your own metrics to track application-specific data, such as processing events or accessing external resources. **Topics** + [ ## How it works ](#monitoring-metrics-custom-howitworks) + [ ## View examples for creating a mapping class ](#monitoring-metrics-custom-examples) + [ ## View custom metrics ](#monitoring-metrics-custom-examples-viewing) ## How it works Custom metrics in Managed Service for Apache Flink use the Apache Flink metric system. Apache Flink metrics have the following attributes: + **Type:** A metric's type describes how it measures and reports data. Available Apache Flink metric types include Count, Gauge, Histogram, and Meter. For more information about Apache Flink metric types, see [Metric Types](https://nightlies.apache.org/flink/flink-docs-release-1.15/monitoring/metrics.html#metric-types). **Note** AWS CloudWatch Metrics does not support the Histogram Apache Flink metric type. CloudWatch can only display Apache Flink metrics of the Count, Gauge, and Meter types. + **Scope:** A metric's scope consists of its identifier and a set of key-value pairs that indicate how the metric will be reported to CloudWatch. A metric's identifier consists of the following: + A system scope, which indicates the level at which the metric is reported (e.g. Operator). + A user scope, that defines attributes such as user variables or the metric group names. These attributes are defined using [https://ci.apache.org/projects/flink/flink-docs-master/api/java/org/apache/flink/metrics/MetricGroup.html#addGroup-java.lang.String-java.lang.String-](https://ci.apache.org/projects/flink/flink-docs-master/api/java/org/apache/flink/metrics/MetricGroup.html#addGroup-java.lang.String-java.lang.String-) or [https://ci.apache.org/projects/flink/flink-docs-master/api/java/org/apache/flink/metrics/MetricGroup.html#addGroup-java.lang.String-](https://ci.apache.org/projects/flink/flink-docs-master/api/java/org/apache/flink/metrics/MetricGroup.html#addGroup-java.lang.String-). For more information about metric scope, see [Scope](https://nightlies.apache.org/flink/flink-docs-release-1.15/monitoring/metrics.html#scope). For more information about Apache Flink metrics, see [Metrics](https://nightlies.apache.org/flink/flink-docs-release-1.15/monitoring/metrics.html) in the [Apache Flink documentation](https://nightlies.apache.org/flink/flink-docs-release-1.15/). To create a custom metric in your Managed Service for Apache Flink, you can access the Apache Flink metric system from any user function that extends `RichFunction` by calling [https://nightlies.apache.org/flink/flink-docs-release-1.15/api/java/org/apache/flink/api/common/functions/RuntimeContext.html#getMetricGroup--](https://nightlies.apache.org/flink/flink-docs-release-1.15/api/java/org/apache/flink/api/common/functions/RuntimeContext.html#getMetricGroup--). This method returns a [MetricGroup](https://nightlies.apache.org/flink/flink-docs-release-1.15/api/java/org/apache/flink/metrics/MetricGroup.html) object you can use to create and register custom metrics. Managed Service for Apache Flink reports all metrics created with the group key `KinesisAnalytics` to CloudWatch. Custom metrics that you define have the following characteristics: + Your custom metric has a metric name and a group name. These names must consist of alphanumeric characters according to [Prometheus naming rules](https://prometheus.io/docs/instrumenting/writing_exporters/#naming). + Attributes that you define in user scope (except for the `KinesisAnalytics` metric group) are published as CloudWatch dimensions. + Custom metrics are published at the `Application` level by default. + Dimensions (Task/ Operator/ Parallelism) are added to the metric based on the application's monitoring level. You set the application's monitoring level using the [MonitoringConfiguration](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_MonitoringConfiguration.html) parameter of the [CreateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_CreateApplication.html) action, or the or [MonitoringConfigurationUpdate](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_MonitoringConfigurationUpdate.html) parameter of the [UpdateApplication](https://docs.aws.amazon.com/managed-flink/latest/apiv2/API_UpdateApplication.html) action. ## View examples for creating a mapping class The following code examples demonstrate how to create a mapping class that creates and increments a custom metric, and how to implement the mapping class in your application by adding it to a `DataStream` object. ### Record count custom metric The following code example demonstrates how to create a mapping class that creates a metric that counts records in a data stream (the same functionality as the `numRecordsIn` metric): ``` private static class NoOpMapperFunction extends RichMapFunction { 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) () -> valueToExpose); } @Override public String map(String value) throws Exception { valueToExpose++; return value; } } ``` In the preceding example, the `valueToExpose` variable is incremented for each record that the application processes. After defining your mapping class, you then create an in-application stream that implements the map: ``` DataStream noopMapperFunctionAfterFilter = kinesisProcessed.map(new NoOpMapperFunction("FilteredRecords")); ``` For the complete code for this application, see [Record Count Custom Metric Application](https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/tree/main/java/CustomMetrics/RecordCount). ### Word count custom metric The following code example demonstrates how to create a mapping class that creates a metric that counts words in a data stream: ``` private static final class Tokenizer extends RichFlatMapFunction> { 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>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)); } } } } ``` In the preceding example, the `counter` variable is incremented for each word that the application processes. After defining your mapping class, you then create an in-application stream that implements the map: ``` // 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> 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()); ``` For the complete code for this application, see [Word Count Custom Metric Application](https://github.com/aws-samples/amazon-managed-service-for-apache-flink-examples/tree/main/java/CustomMetrics/WordCount). ## View custom metrics Custom metrics for your application appear in the CloudWatch Metrics console in the **AWS/KinesisAnalytics** dashboard, under the **Application** metric group. # Use CloudWatch Alarms with Amazon Managed Service for Apache Flink Using Amazon CloudWatch metric alarms, you watch a CloudWatch metric over a time period that you specify. The alarm performs one or more actions based on the value of the metric or expression relative to a threshold over a number of time periods. An example of an action is sending a notification to an Amazon Simple Notification Service (Amazon SNS) topic. For more information about CloudWatch alarms, see [Using Amazon CloudWatch Alarms](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/AlarmThatSendsEmail.html). ## Review recommended alarms This section contains the recommended alarms for monitoring Managed Service for Apache Flink applications. The table describes the recommended alarms and has the following columns: + **Metric Expression:** The metric or metric expression to test against the threshold. + **Statistic:** The statistic used to check the metric—for example, **Average**. + **Threshold:** Using this alarm requires you to determine a threshold that defines the limit of expected application performance. You need to determine this threshold by monitoring your application under normal conditions. + **Description:** Causes that might trigger this alarm, and possible solutions for the condition. | Metric Expression | Statistic | Threshold | Description | | --- |--- |--- |--- | | downtime > 0 | Average | 0 | A downtime greater than zero indicates that the application has failed. If the value is larger than 0, the application is not processing any data. Recommended for all applications. The Downtime metric measures the duration of an outage. A downtime greater than zero indicates that the application has failed. For troubleshooting, see [Application is restarting](troubleshooting-rt-restarts.md). | | RATE (numberOfFailedCheckpoints) > 0 | Average | 0 | This metric counts the number of failed checkpoints since the application started. Depending on the application, it can be tolerable if checkpoints fail occasionally. But if checkpoints are regularly failing, the application is likely unhealthy and needs further attention. We recommend monitoring RATE(numberOfFailedCheckpoints) to alarm on the gradient and not on absolute values. Recommended for all applications. Use this metric to monitor application health and checkpointing progress. The application saves state data to checkpoints when it's healthy. Checkpointing can fail due to timeouts if the application isn't making progress in processing the input data. For troubleshooting, see [Checkpointing is timing out](troubleshooting-chk-timeout.md). | | Operator.numRecordsOutPerSecond < threshold | Average | The minimum number of records emitted from the application during normal conditions. | Recommended for all applications. Falling below this threshold can indicate that the application isn't making expected progress on the input data. For troubleshooting, see [Throughput is too slow](troubleshooting-rt-throughput.md). | | records\$1lag\$1max\$1millisbehindLatest > threshold | Maximum | The maximum expected latency during normal conditions. | If the application is consuming from Kinesis or Kafka, these metrics indicate if the application is falling behind and needs to be scaled in order to keep up with the current load. This is a good generic metric that is easy to track for all kinds of applications. But it can only be used for reactive scaling, i.e., when the application has already fallen behind. Recommended for all applications. Use the records\$1lag\$1max metric for a Kafka source, or the millisbehindLatest for a Kinesis stream source. Rising above this threshold can indicate that the application isn't making expected progress on the input data. For troubleshooting, see [Throughput is too slow](troubleshooting-rt-throughput.md). | | lastCheckpointDuration > threshold | Maximum | The maximum expected checkpoint duration during normal conditions. | Monitors how much data is stored in state and how long it takes to take a checkpoint. If checkpoints grow or take long, the application is continuously spending time on checkpointing and has less cycles for actual processing. At some points, checkpoints may grow too large or take so long that they fail. In addition to monitoring absolute values, customers should also considering monitoring the change rate with RATE(lastCheckpointSize) and RATE(lastCheckpointDuration). If the lastCheckpointDuration continuously increases, rising above this threshold can indicate that the application isn't making expected progress on the input data, or that there are problems with application health such as backpressure. For troubleshooting, see [Unbounded state growth](troubleshooting-rt-stateleaks.md). | | lastCheckpointSize > threshold | Maximum | The maximum expected checkpoint size during normal conditions. | Monitors how much data is stored in state and how long it takes to take a checkpoint. If checkpoints grow or take long, the application is continuously spending time on checkpointing and has less cycles for actual processing. At some points, checkpoints may grow too large or take so long that they fail. In addition to monitoring absolute values, customers should also considering monitoring the change rate with RATE(lastCheckpointSize) and RATE(lastCheckpointDuration). If the lastCheckpointSize continuously increases, rising above this threshold can indicate that the application is accumulating state data. If the state data becomes too large, the application can run out of memory when recovering from a checkpoint, or recovering from a checkpoint might take too long. For troubleshooting, see [Unbounded state growth](troubleshooting-rt-stateleaks.md). | | heapMemoryUtilization > threshold | Maximum | This gives a good indication of the overall resource utilization of the application and can be used for proactive scaling unless the application is I/O bound. The maximum expected heapMemoryUtilization size during normal conditions, with a recommended value of 90 percent. | You can use this metric to monitor the maximum memory utilization of task managers across the application. If the application reaches this threshold, you need to provision more resources. You do this by enabling automatic scaling or increasing the application parallelism. For more information about increasing resources, see [Implement application scaling](how-scaling.md). | | cpuUtilization > threshold | Maximum | This gives a good indication of the overall resource utilization of the application and can be used for proactive scaling unless the application is I/O bound. The maximum expected cpuUtilization size during normal conditions, with a recommended value of 80 percent. | You can use this metric to monitor the maximum CPU utilization of task managers across the application. If the application reaches this threshold, you need to provision more resources You do this by enabling automatic scaling or increasing the application parallelism. For more information about increasing resources, see [Implement application scaling](how-scaling.md). | | threadsCount > threshold | Maximum | The maximum expected threadsCount size during normal conditions. | You can use this metric to watch for thread leaks in task managers across the application. If this metric reaches this threshold, check your application code for threads being created without being closed. | | (oldGarbageCollectionTime \$1 100)/60\$1000 over 1 min period') > threshold | Maximum | The maximum expected oldGarbageCollectionTime duration. We recommend setting a threshold such that typical garbage collection time is 60 percent of the specified threshold, but the correct threshold for your application will vary. | If this metric is continually increasing, this can indicate that there is a memory leak in task managers across the application. | | RATE(oldGarbageCollectionCount) > threshold | Maximum | The maximum expected oldGarbageCollectionCount under normal conditions. The correct threshold for your application will vary. | If this metric is continually increasing, this can indicate that there is a memory leak in task managers across the application. | | Operator.currentOutputWatermark - Operator.currentInputWatermark > threshold | Minimum | The minimum expected watermark increment under normal conditions. The correct threshold for your application will vary. | If this metric is continually increasing, this can indicate that either the application is processing increasingly older events, or that an upstream subtask has not sent a watermark in an increasingly long time. |