# Flink支持Metrics上报Kafka
#### 使用场景
Flink提供了监控指标（Metrics）来帮助分析作业，该特性支持将配置的Metrics上报到Kafka中，便于用户通过Kafka消费指标数据进行实时监控和分析。
#### 前提条件
- Flink集群与Metrics上报的Kafka网络连通。
- Metrics上报的Kafka Topic需要提前创建好。
 
#### 约束与限制
本章节适用于MRS 3.6.0-LTS及之后的版本。
当前只支持使用Kerberos方式对接Kafka，不支持用户名密码的方式对接Kafka。
#### 配置步骤
1. 配置Flink支持Metrics上报Kafka参数。 
   **通过FlinkServer提交作业，Metrics上报Kafka。**
   - 在Flink配置界面配置Flink支持Metrics上报Kafka参数
     1. 登录Manager页面，选择"集群 \> 服务 \> Flink \> 配置 \> 全部配置"。
     
     2. 选择"FlinkServer（角色） \> 自定义"，在"flink.customized.configs"参数中添加[表1]中的参数，添加完成后单击"保存"。
     
     3. 重启受影响的FlinkServer实例。
      
   
   - 在FlinkServer作业开发界面配置Flink支持Metrics上报Kafka参数
     1. 选择"集群 \> 服务 \> Flink"，在"Flink WebUI"右侧，单击链接，访问Flink的WebUI。
     
     2. 单击"作业管理"进入作业管理页面。
     
     3. 在指定作业"操作"列单击"开发"进入作业开发界面。
     
     4. 在作业开发界面的"自定义参数"项中添加[表1]中的参数，添加完成后单击"保存"。
      
   
   
   **通过客户端提交作业，Metrics上报Kafka。**
   1. 登录Manager页面，选择"集群 \> 服务 \> Flink \> 配置 \> 全部配置"。
   
   2. 选择"FlinkResource（角色） \> 自定义"，在"flink.customized.configs"参数中添加[表1]中的参数，添加完成后单击"保存"，并重启受影响的FlinkResource实例。
   
   3. 重新安装客户端或更新客户端配置，具体请参考[安装客户端](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_2127.html)或[更新客户端](https://support.huaweicloud.com/cmpntguide-lts-mrs/mrs_01_2129.html)。
   
   
    表1Flink支持Metrics上报Kafka参数配置 
   | 名称                                          | 值                                                                                     | 是否必填 | 描述                                                                                                                                                                                                                                                                                                                                                                                                                                               |
   |:---|:---|:---|:---|
   | metrics.reporter                            | alarm,kafka                                                                           | 是    | 新增的reporter需要追加对应名称。alarm用于Flinkserver的告警，防止自定义参数覆盖Flink原有的参数，需同时填写alarm和kafka。                                                                                                                                                                                                                                                                                                                                                                  |
   | metrics.reporter.kafka.factory.class        | com.huawei.flink.metrics.kafka.KafkaReporterFactory                                   | 是    | Flink Metric指标上报给kafka的实现类，值固定为： com.huawei.flink.metrics.kafka.KafkaReporterFactory。                                                                                                                                                                                                                                                                                              |
   | metrics.reporter.kafka.interval             | 30s                                                                                   | 是    | Flink Metric指标上报kafka的周期，单位：秒。                                                                                                                                                                                                                                                                                                                                                                                                                   |
   | metrics.reporter.kafka.bootstrapServers     | Kafka集群IP1:端口号,Kafka集群IP2:端口号                                                         | 是    | Flink Metric指标上报kafka的broker IP、端口。                                                                                                                                                                                                                                                                                                                                                                                                              |
   | metrics.reporter.kafka.metricNames          | -                                                                                     | 是    | 允许上报的监控项。如：numRecordsIn,numRecordsOut。                                                                                                                                                                                                                                                                                                                                                                                                           |
   | metrics.reporter.kafka.topicName            | -                                                                                     | 是    | 上报给Kafka的Topic，需要提前创建好。如：flink-metrics。                                                                                                                                                                                                                                                                                                                                                                                                          |
   |  metrics.reporter.kafka.enableJobInfo | 取值范围：true/false 默认值：false | 否    | 是否在指标上报中添加作业信息，包括作业名、提交用户以及Yarn队列名。填入true时开启。                                                                                                                                                                                                                                                                                                                                                                                                    |
   | metrics.reporter.kafka.yarnQueue            | -                                                                                     | 否    | 上报指标JSON中添加的Yarn队列名。                                                                                                                                                                                                                                                                                                                                                                                                                             |
   | metrics.reporter.kafka.jobName              | -                                                                                     | 否    | 上报指标JSON中添加的作业名。                                                                                                                                                                                                                                                                                                                                                                                                                                 |
   | metrics.reporter.kafka.properties.kafka配置参数 | -                                                                                     | 否    | 如果有其他Kafka参数需要配置，可以通过此参数填写。例如认证参数（安全模式集群必填）： metrics.reporter.kafka.properties.security.protocol=SASL_PLAINTEXT metrics.reporter.kafka.properties.sasl.kerberos.service.name=kafka metrics.reporter.kafka.properties.kerberos.domain.name=hadoop.xxxx.COM |
      
   
   
2. 提交Flink作业，对应的Metrics就会被上报到配置的Kafka Topic中。 
   数据格式参考如下JSON：
   ```
   {
    "jobName": "kafka_test",
    "submitUserName": "xxx",
    "metric": {
     "tm_id": "container_xxx",
     "host": "x-x-x-x",
     "name": "flink_taskmanager.Status.JVM.Memory.NonHeap_Used",
     "metricType": "GAUGE",
        "value": 120661112
    },
    "reportTime": 1761530081855,
    "yarnQueue": "default"
   }
   ```
   ![](https://support.huaweicloud.com/cmpntguide-lts-mrs/public_sys-resources/warning_3.0-zh-cn.png)
   如果配置的Kafka服务出现异常，无法连接时，Flink作业业务不会中断，但是Metrics功能将会受到影响，监控指标无法发送给Kafka。