47、Flink 的指标报告介绍(graphite、influxdb、prometheus、statsd和datalog)及示例(jmx和slf4j示例)

Flink 系列文章

1、Flink 部署、概念介绍、source、transformation、sink使用示例、四大基石介绍和示例等系列综合文章链接

13、Flink 的table api与sql的基本概念、通用api介绍及入门示例
14、Flink 的table api与sql之数据类型: 内置数据类型以及它们的属性
15、Flink 的table api与sql之流式概念-详解的介绍了动态表、时间属性配置(如何处理更新结果)、时态表、流上的join、流上的确定性以及查询配置
16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及FileSystem示例(1)
16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及Elasticsearch示例(2)
16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及Apache Kafka示例(3)
16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及JDBC示例(4)
16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及Apache Hive示例(6)
17、Flink 之Table API: Table API 支持的操作(1)
17、Flink 之Table API: Table API 支持的操作(2)
18、Flink的SQL 支持的操作和语法
19、Flink 的Table API 和 SQL 中的内置函数及示例(1)
19、Flink 的Table API 和 SQL 中的自定义函数及示例(2)
19、Flink 的Table API 和 SQL 中的自定义函数及示例(3)
19、Flink 的Table API 和 SQL 中的自定义函数及示例(4)
20、Flink SQL之SQL Client: 不用编写代码就可以尝试 Flink SQL,可以直接提交 SQL 任务到集群上
21、Flink 的table API与DataStream API 集成(1)- 介绍及入门示例、集成说明
21、Flink 的table API与DataStream API 集成(2)- 批处理模式和inser-only流处理
21、Flink 的table API与DataStream API 集成(3)- changelog流处理、管道示例、类型转换和老版本转换示例
21、Flink 的table API与DataStream API 集成(完整版)
22、Flink 的table api与sql之创建表的DDL
24、Flink 的table api与sql之Catalogs(介绍、类型、java api和sql实现ddl、java api和sql操作catalog)-1
24、Flink 的table api与sql之Catalogs(java api操作数据库、表)-2
24、Flink 的table api与sql之Catalogs(java api操作视图)-3
24、Flink 的table api与sql之Catalogs(java api操作分区与函数)-4
25、Flink 的table api与sql之函数(自定义函数示例)
26、Flink 的SQL之概览与入门示例
27、Flink 的SQL之SELECT (select、where、distinct、order by、limit、集合操作和去重)介绍及详细示例(1)
27、Flink 的SQL之SELECT (SQL Hints 和 Joins)介绍及详细示例(2)
27、Flink 的SQL之SELECT (窗口函数)介绍及详细示例(3)
27、Flink 的SQL之SELECT (窗口聚合)介绍及详细示例(4)
27、Flink 的SQL之SELECT (Group Aggregation分组聚合、Over Aggregation Over聚合 和 Window Join 窗口关联)介绍及详细示例(5)
27、Flink 的SQL之SELECT (Top-N、Window Top-N 窗口 Top-N 和 Window Deduplication 窗口去重)介绍及详细示例(6)
27、Flink 的SQL之SELECT (Pattern Recognition 模式检测)介绍及详细示例(7)
28、Flink 的SQL之DROP 、ALTER 、INSERT 、ANALYZE 语句
29、Flink SQL之DESCRIBE、EXPLAIN、USE、SHOW、LOAD、UNLOAD、SET、RESET、JAR、JOB Statements、UPDATE、DELETE(1)
29、Flink SQL之DESCRIBE、EXPLAIN、USE、SHOW、LOAD、UNLOAD、SET、RESET、JAR、JOB Statements、UPDATE、DELETE(2)
30、Flink SQL之SQL 客户端(通过kafka和filesystem的例子介绍了配置文件使用-表、视图等)
31、Flink的SQL Gateway介绍及示例
32、Flink table api和SQL 之用户自定义 Sources & Sinks实现及详细示例
33、Flink 的Table API 和 SQL 中的时区
35、Flink 的 Formats 之CSV 和 JSON Format
36、Flink 的 Formats 之Parquet 和 Orc Format
41、Flink之Hive 方言介绍及详细示例
40、Flink 的Apache Kafka connector(kafka source的介绍及使用示例)-1
40、Flink 的Apache Kafka connector(kafka sink的介绍及使用示例)-2
40、Flink 的Apache Kafka connector(kafka source 和sink 说明及使用示例) 完整版
42、Flink 的table api与sql之Hive Catalog
43、Flink之Hive 读写及详细验证示例
44、Flink之module模块介绍及使用示例和Flink SQL使用hive内置函数及自定义函数详细示例–网上有些说法好像是错误的
45、Flink 的指标体系介绍及验证(1)-指标类型及指标实现示例
45、Flink 的指标体系介绍及验证(2)-指标的scope、报告、系统指标以及追踪、api集成示例和dashboard集成
45、Flink 的指标体系介绍及验证(3)- 完整版
46、Flink 的table api与sql之配项列表及示例
47、Flink 的指标报告介绍(graphite、influxdb、prometheus、statsd和datalog)及示例(jmx和slf4j示例)


文章目录

  • Flink 系列文章
  • 一、Metric Reporters
    • 1、概述及示例
    • 2、入门示例
      • 0)、特别说明
      • 1)、配置
      • 2)、验证
      • 3)、自定义的指标收集器
    • 3、基于标志符格式 vs. 基于 tags 格式
    • 4、Push vs. Pull
    • 5、发送器
      • 1)、JMX
      • 2)、Graphite
      • 2)、InfluxDB
      • 4)、Prometheus
      • 5)、PrometheusPushGateway
      • 6)、StatsD
      • 7)、Datadog
      • 8)、Slf4j


本文介绍了Flink的指标报告的类型以及几种发送器的示例配置,特别是针对jmx和slf4j给出了具体的实现示例及验证结果。由于本文是介绍flink的指标数据可以给哪些系统发送数据,至于具体实现算是各自的应用,故没有深入的展示。
本文依赖flink集群能正常使用。
本文分为5个部分,即Metric Reporters概述、入门示例、格式、推拉和8种发送器,其中两种给出了具体的示例。
本文的示例是在Flink 1.17版本中运行。

一、Metric Reporters

1、概述及示例

Flink 支持用户将 Flink 的各项运行时指标发送给外部系统。 了解更多指标方面信息可查看 45、Flink 的指标体系介绍及验证(3)- 完整版。

你可以通过 conf/flink-conf.yaml 文件来配置一种或多种发送器,将运行时指标暴露给外部系统。 发送器会在 TaskManager、Flink 作业启动时进行实例化。

下面列出了所有发送器都适用的参数,可以通过配置文件中的 metrics.reporter.<reporter_name>. 项进行配置。有些发送器有自己特有的配置,详见该发送器章节下的具体说明。
在这里插入图片描述
每种发送器的配置需要有 factory.class 属性。 有些基于定时调度的发送器还可以通过 interval 来配置发送间隔。

Flink 在启动时必须能访问到发送器所属的 jar 包,发送器会被加载为 plugins,Flink 自带的发送器(文档中已经列出的发送器)无需做其他配置,开箱即用。

[root@1049 plugins]# ll
drwxr-xr-x 2 501 games 114 519 2023 external-resource-gpu
drwxr-xr-x 2 501 games  46 519 2023 metrics-datadog
drwxr-xr-x 2 501 games  47 519 2023 metrics-graphite
drwxr-xr-x 2 501 games  47 519 2023 metrics-influx
drwxr-xr-x 2 501 games  42 519 2023 metrics-jmx
drwxr-xr-x 2 501 games  49 519 2023 metrics-prometheus
drwxr-xr-x 2 501 games  44 519 2023 metrics-slf4j
drwxr-xr-x 2 501 games  45 519 2023 metrics-statsd

你可以实现 org.apache.flink.metrics.reporter.MetricReporter 接口来自定义发送器,并实现 Scheduled 接口让发送器周期性地将运行时指标发送出去。 需要注意 report() 方法不应该阻塞太长的时间,所有用时很长的操作应该异步执行。 另外也可以实现 MetricReporterFactory 接口,让发送器作为插件被 Flink 导入。

2、入门示例

想要同时配置多个发送器,可参考以下示例。

0)、特别说明

  • 如果要使用jconsole远程连接flink,必须在conf/flink-conf.yaml配置文件中增加如下内容,具体情况根据自己的环境进行修改
  • 如果自己的flink集群环境既包含job manager,又包含task manager,则上面的端口配置会出现冲突,一般来说应该是不需要,因为一台机器有一个监控即可。
  • 1099端口可以根据自己的需要视情况设置
env.java.opts: "-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.local.only=false -Dcom.sun.management.jmxremote.port=1099 -Dcom.sun.management.jmxremote.rmi.port=1099 -Djava.rmi.server.hostname=192.168.10.49"

1)、配置

本示例配置将指标写入jmx和日志中(将指标每分钟写入一次日志文件中)

metrics.reporters: alan_jmx_reporter,alan_slf4j_reporter

metrics.reporter.alan_jmx_reporter.factory.class: org.apache.flink.metrics.jmx.JMXReporterFactory
metrics.reporter.alan_jmx_reporter.port: 9020-9040
metrics.reporter.alan_jmx_reporter.scope.variables.excludes: job_id;task_attempt_num
metrics.reporter.alan_jmx_reporter.scope.variables.additional: cluster_name:alan_test_cluster,tag_name:alan_tag_value

metrics.reporter.alan_slf4j_reporter.factory.class: org.apache.flink.metrics.slf4j.Slf4jReporterFactory
metrics.reporter.alan_slf4j_reporter.interval: 60 SECONDS

2)、验证

  • 日志文件内容
    包含job manager和task manager,示例如下
2023-11-28 14:22:40,280 INFO  org.apache.flink.metrics.slf4j.Slf4jReporter                 [] - 
=========================== Starting metrics report ===========================

-- Gauges ---------------------------------------------------------------------
192_168_10_49.jobmanager.Status.JVM.ClassLoader.ClassesLoaded: 9047
192_168_10_49.jobmanager.Status.JVM.Memory.Mapped.MemoryUsed: 0
192_168_10_49.jobmanager.Status.JVM.Memory.NonHeap.Committed: 82313216
192_168_10_49.jobmanager.numRunningJobs: 0
192_168_10_49.jobmanager.Status.JVM.Memory.Metaspace.Used: 58573704
192_168_10_49.jobmanager.Status.JVM.Memory.Mapped.Count: 0
192_168_10_49.jobmanager.Status.JVM.ClassLoader.ClassesUnloaded: 0
192_168_10_49.jobmanager.Status.JVM.Memory.Heap.Used: 69453280
192_168_10_49.jobmanager.Status.JVM.CPU.Load: 2.887393783748263E-7
192_168_10_49.jobmanager.Status.JVM.GarbageCollector.G1 Young Generation.Count: 3
192_168_10_49.jobmanager.Status.JVM.Memory.Metaspace.Committed: 60817408
192_168_10_49.jobmanager.Status.JVM.GarbageCollector.G1 Young Generation.Time: 27
192_168_10_49.jobmanager.numRegisteredTaskManagers: 0
192_168_10_49.jobmanager.Status.JVM.GarbageCollector.G1 Old Generation.Count: 0
192_168_10_49.jobmanager.Status.JVM.GarbageCollector.G1 Old Generation.Time: 0
192_168_10_49.jobmanager.Status.JVM.Memory.Direct.Count: 10
192_168_10_49.jobmanager.Status.JVM.Memory.Metaspace.Max: 268435456
192_168_10_49.jobmanager.taskSlotsTotal: 0
192_168_10_49.jobmanager.Status.JVM.Memory.NonHeap.Used: 76532504
192_168_10_49.jobmanager.Status.JVM.Memory.Heap.Committed: 1530920960
192_168_10_49.jobmanager.Status.JVM.Memory.NonHeap.Max: 780140544
192_168_10_49.jobmanager.Status.JVM.Memory.Direct.MemoryUsed: 532481
192_168_10_49.jobmanager.taskSlotsAvailable: 0
192_168_10_49.jobmanager.Status.JVM.Memory.Heap.Max: 1530920960
192_168_10_49.jobmanager.Status.JVM.Memory.Direct.TotalCapacity: 532480
192_168_10_49.jobmanager.Status.JVM.Memory.Mapped.TotalCapacity: 0
192_168_10_49.jobmanager.Status.JVM.Threads.Count: 69
192_168_10_49.jobmanager.Status.JVM.CPU.Time: 9960000000

=========================== Finished metrics report ===========================

2023-11-28 14:23:40,281 INFO  org.apache.flink.metrics.slf4j.Slf4jReporter                 [] - 
=========================== Starting metrics report ===========================

-- Gauges ---------------------------------------------------------------------
192_168_10_49.jobmanager.Status.JVM.ClassLoader.ClassesLoaded: 9121
192_168_10_49.jobmanager.Status.JVM.Memory.Mapped.MemoryUsed: 0
192_168_10_49.jobmanager.Status.JVM.Memory.NonHeap.Committed: 83361792
192_168_10_49.jobmanager.numRunningJobs: 0
192_168_10_49.jobmanager.Status.JVM.Memory.Metaspace.Used: 59044888
192_168_10_49.jobmanager.Status.JVM.Memory.Mapped.Count: 0
192_168_10_49.jobmanager.Status.JVM.ClassLoader.ClassesUnloaded: 0
192_168_10_49.jobmanager.Status.JVM.Memory.Heap.Used: 73647584
192_168_10_49.jobmanager.Status.JVM.CPU.Load: 0.005763448045439359
192_168_10_49.jobmanager.Status.JVM.GarbageCollector.G1 Young Generation.Count: 3
192_168_10_49.jobmanager.Status.JVM.Memory.Metaspace.Committed: 61472768
192_168_10_49.jobmanager.Status.JVM.GarbageCollector.G1 Young Generation.Time: 27
192_168_10_49.jobmanager.numRegisteredTaskManagers: 0
192_168_10_49.jobmanager.Status.JVM.GarbageCollector.G1 Old Generation.Count: 0
192_168_10_49.jobmanager.Status.JVM.GarbageCollector.G1 Old Generation.Time: 0
192_168_10_49.jobmanager.Status.JVM.Memory.Direct.Count: 10
192_168_10_49.jobmanager.Status.JVM.Memory.Metaspace.Max: 268435456
192_168_10_49.jobmanager.taskSlotsTotal: 0
192_168_10_49.jobmanager.Status.JVM.Memory.NonHeap.Used: 78078760
192_168_10_49.jobmanager.Status.JVM.Memory.Heap.Committed: 1530920960
192_168_10_49.jobmanager.Status.JVM.Memory.NonHeap.Max: 780140544
192_168_10_49.jobmanager.Status.JVM.Memory.Direct.MemoryUsed: 532481
192_168_10_49.jobmanager.taskSlotsAvailable: 0
192_168_10_49.jobmanager.Status.JVM.Memory.Heap.Max: 1530920960
192_168_10_49.jobmanager.Status.JVM.Memory.Direct.TotalCapacity: 532480
192_168_10_49.jobmanager.Status.JVM.Memory.Mapped.TotalCapacity: 0
192_168_10_49.jobmanager.Status.JVM.Threads.Count: 63
192_168_10_49.jobmanager.Status.JVM.CPU.Time: 11330000000

=========================== Finished metrics report ===========================

。。。。。。。。。。。。。。

2023-11-28 13:42:26,246 INFO  org.apache.flink.metrics.slf4j.Slf4jReporter                 [] - 
=========================== Starting metrics report ===========================

-- Gauges ---------------------------------------------------------------------
localhost.taskmanager.localhost:46098-60172b.Status.JVM.Memory.Metaspace.Committed: 54468608
localhost.taskmanager.localhost:46098-60172b.Status.Shuffle.Netty.TotalMemory: 359694336
localhost.taskmanager.localhost:46098-60172b.Status.JVM.CPU.Time: 9470000000
localhost.taskmanager.localhost:46098-60172b.Status.JVM.ClassLoader.ClassesLoaded: 7916
localhost.taskmanager.localhost:46098-60172b.Status.JVM.Threads.Count: 51
localhost.taskmanager.localhost:46098-60172b.Status.Network.AvailableMemorySegments: 10977
localhost.taskmanager.localhost:46098-60172b.Status.JVM.GarbageCollector.G1 Old Generation.Count: 0
localhost.taskmanager.localhost:46098-60172b.Status.JVM.Memory.Mapped.Count: 0
localhost.taskmanager.localhost:46098-60172b.Status.JVM.Memory.Mapped.MemoryUsed: 0
localhost.taskmanager.localhost:46098-60172b.Status.JVM.Memory.NonHeap.Committed: 73748480
localhost.taskmanager.localhost:46098-60172b.Status.JVM.GarbageCollector.G1 Young Generation.Count: 3
localhost.taskmanager.localhost:46098-60172b.Status.JVM.CPU.Load: 2.748870782605016E-7
localhost.taskmanager.localhost:46098-60172b.Status.JVM.Memory.Metaspace.Max: 268435456
localhost.taskmanager.localhost:46098-60172b.Status.JVM.GarbageCollector.G1 Young Generation.Time: 63
localhost.taskmanager.localhost:46098-60172b.Status.JVM.Memory.Mapped.TotalCapacity: 0
localhost.taskmanager.localhost:46098-60172b.Status.Flink.Memory.Managed.Total: 1438814063
localhost.taskmanager.localhost:46098-60172b.Status.Network.TotalMemorySegments: 10977
localhost.taskmanager.localhost:46098-60172b.Status.JVM.Memory.Heap.Committed: 1665138688
localhost.taskmanager.localhost:46098-60172b.Status.JVM.Memory.Direct.TotalCapacity: 360426496
localhost.taskmanager.localhost:46098-60172b.Status.Shuffle.Netty.AvailableMemory: 359694336
localhost.taskmanager.localhost:46098-60172b.Status.JVM.Memory.Direct.MemoryUsed: 360426497
localhost.taskmanager.localhost:46098-60172b.Status.JVM.Memory.Heap.Used: 38173672
localhost.taskmanager.localhost:46098-60172b.Status.JVM.Memory.Metaspace.Used: 52683336
localhost.taskmanager.localhost:46098-60172b.Status.JVM.GarbageCollector.G1 Old Generation.Time: 0
localhost.taskmanager.localhost:46098-60172b.Status.Flink.Memory.Managed.Used: 0
localhost.taskmanager.localhost:46098-60172b.Status.Shuffle.Netty.RequestedMemoryUsage: 0
localhost.taskmanager.localhost:46098-60172b.Status.JVM.Memory.NonHeap.Used: 69118632
localhost.taskmanager.localhost:46098-60172b.Status.Shuffle.Netty.TotalMemorySegments: 10977
localhost.taskmanager.localhost:46098-60172b.Status.Shuffle.Netty.UsedMemorySegments: 0
localhost.taskmanager.localhost:46098-60172b.Status.JVM.Memory.Heap.Max: 1665138688
localhost.taskmanager.localhost:46098-60172b.Status.Shuffle.Netty.AvailableMemorySegments: 10977
localhost.taskmanager.localhost:46098-60172b.Status.Shuffle.Netty.UsedMemory: 0
localhost.taskmanager.localhost:46098-60172b.Status.JVM.ClassLoader.ClassesUnloaded: 0
localhost.taskmanager.localhost:46098-60172b.Status.JVM.Memory.Direct.Count: 10993
localhost.taskmanager.localhost:46098-60172b.Status.JVM.Memory.NonHeap.Max: 780140544

=========================== Finished metrics report ===========================

  • jconsole显示
    在这里插入图片描述

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

3)、自定义的指标收集器

1、自定义指标参考文章:45、Flink 的指标体系介绍及验证(1)-指标类型及指标实现示例

2、将实现的自定义指标打包,并放在flink的lib目录下,确保Flink 在启动时必须能访问到发送器所属的 jar 包,发送器会被加载为 plugins

3、配置flink-conf.yaml文件

metrics.reporters: alan-reporter
metrics.reporter.alan-reporter.class: com.alan.AlanReporter
metrics.reporter.alan-reporter.class.host: 192.168.10.49
metrics.reporter.alan-reporter.port: 8789

一般而言,不需要自己开发,除非有特殊需求一般的开源不能满足。

3、基于标志符格式 vs. 基于 tags 格式

发送器发送运行指标时有以下两种格式:

第一种是基于标志符的格式,这种格式将指标所属的详细范围信息与该指标名称连在一起,组成一个长字符串标志符。 比如 job.MyJobName.numRestarts 就是这样的格式。

第二种是基于 tags 的格式,这种格式由指标的逻辑范围和该指标名称组成,代表某一类通用的指标,比如 job.numRestarts。 这类格式描述的特定指标实例会以“键值对”的方式体现出来,即所谓的标签或变量,比如 “jobName=MyJobName”。

4、Push vs. Pull

发送器通过 Pull 或 Push 的方式将指标发送出去。
基于 Push 的发送器一般实现了 Scheduled 接口,周期性地将当前运行指标信息汇总,发送给外部系统存储。
基于 Pull 的发送器一般是由外部系统主动发起查询将指标信息拉走。

5、发送器

接下来的部分列出了 Flink 支持的发送器。

1)、JMX

参考本文的入门示例,不再赘述。

  • factory.class
metrics.reporter.alan_jmx_reporter.factory.class: org.apache.flink.metrics.jmx.JMXReporterFactory
  • 类型: pull/基于 tags 格式

  • 参数:
    port -(可选的)JMX 监听的端口。 如果需要在一台机器上运行多个发送器示例进行监控时(比如 TaskManger 与 JobManager 在一台机器上运行时),建议将端口号配置为 9250-9260 这样的区间, 实际使用的端口会在相关作业 或 TaskManger 的日志中显示。如果设置了这个选项,Flink 会按照配置的端口号或端口区间开启 JMX 发送器, 这些运行时指标可以通过本地的 JMX 默认接口访问到。

metrics.reporter.jmx.factory.class: org.apache.flink.metrics.jmx.JMXReporterFactory
metrics.reporter.jmx.port: 8789
# 或,下面例子重命名/并增加了其他属性配置
metrics.reporters: alan_jmx_reporter,alan_slf4j_reporter
metrics.reporter.alan_jmx_reporter.factory.class: org.apache.flink.metrics.jmx.JMXReporterFactory
metrics.reporter.alan_jmx_reporter.port: 9020-9040
metrics.reporter.alan_jmx_reporter.scope.variables.excludes: job_id;task_attempt_num
metrics.reporter.alan_jmx_reporter.scope.variables.additional: cluster_name:alan_test_cluster,tag_name:alan_tag_value

通过 JMX 发送的指标由“域”和“key-properties”列表组成,“域”总以 org.apache.flink 开头,后跟一个“通用指标标识符”。

“通用指标标识符”不像一般系统的指标度量那样按照所度量范围的形式进行命名,而是不包含任何变量,如同常量一样,每个作业都有。 例如,org.apache.flink.job.task.numBytesOut 就是一个“域”。

“key-property”列表包含所有指标的具体变量值,无论配置的度量范围格式如何,都能关联到指定的指标。

例如host=localhost,job_name=MyJob,task_name=MyTask 就是一个“key-property”列表。

总结来说,“域”标注出了某个指标的类,“key-property”列表标注出了该指标的一个(或者多个)实例。

2)、Graphite

关于Graphite 请参考官方文档,不再赘述。

  • factory.class
metrics.reporter.grph.factory.class: org.apache.flink.metrics.graphite.GraphiteReporterFactory
  • 类型:push/基于标志符格式

  • 参数:
    host - Graphite 服务的地址。
    port - Graphite 服务的端口。
    protocol - 使用的协议(TCP/UDP)。

  • 配置示例:

metrics.reporter.grph.factory.class: org.apache.flink.metrics.graphite.GraphiteReporterFactory
metrics.reporter.grph.host: localhost
metrics.reporter.grph.port: 2003
metrics.reporter.grph.protocol: TCP
metrics.reporter.grph.interval: 60 SECONDS

2)、InfluxDB

关于 InfluxDB 请参考其官网介绍。

  • factory.class
metrics.reporter.influxdb.factory.class: org.apache.flink.metrics.influxdb.InfluxdbReporterFactory
  • 类型: push/基于 tags 格式

  • 参数:
    在这里插入图片描述

  • 配置示例:

metrics.reporter.influxdb.factory.class: org.apache.flink.metrics.influxdb.InfluxdbReporterFactory
metrics.reporter.influxdb.scheme: http
metrics.reporter.influxdb.host: localhost
metrics.reporter.influxdb.port: 8086
metrics.reporter.influxdb.db: flink
metrics.reporter.influxdb.username: flink-metrics
metrics.reporter.influxdb.password: qwerty
metrics.reporter.influxdb.retentionPolicy: one_hour
metrics.reporter.influxdb.consistency: ANY
metrics.reporter.influxdb.connectTimeout: 60000
metrics.reporter.influxdb.writeTimeout: 60000
metrics.reporter.influxdb.interval: 60 SECONDS

InfluxDB 发送器会使用 http 协议按照将指标发送到 InfluxDB 服务器。指标的保留策略可通过配置指定,或按照 InfluxDB 服务端的保留策略决定。 所有的 Flink 运行指标变量(见 List of all Variables)都会按照 tag 形式上报给 InfluxDB。

4)、Prometheus

关于Prometheus请参考官网信息。

  • factory.class
metrics.reporter.prom.factory.class: org.apache.flink.metrics.prometheus.PrometheusReporterFactory
  • 类型: pull/基于 tags 格式

  • 参数:
    1、port - (可选的)Prometheus 发送器监听的端口,默认为 9249。如果需要在一台机器上运行多个发送器示例进行监控时(比如 TaskManger 与 JobManager 在一台机器上运行时),建议将端口号配置为 9250-9260 这样的区间。
    2、filterLabelValueCharacters -(可选的)指定是否过滤 label 中的非法字符。如果设置过滤,所有没有按照 [a-zA-Z0-9:_] 匹配的字符都会被过滤掉,如果设置不过滤,则不会有字符被过滤掉。设置不过滤前,请确保你的 label 符合 Prometheus 的 lable 命名规范。

  • 配置示例:

metrics.reporter.prom.factory.class: org.apache.flink.metrics.prometheus.PrometheusReporterFactory

Flink 指标类型与 Prometheus 指标类型对应关系如下:
在这里插入图片描述
所有的 Flink 运行指标变量(见 List of all Variables)都会按照 label 形式上报给 Prometheus。

5)、PrometheusPushGateway

关于PrometheusPushGateway请参考官网信息。

  • factory.class
metrics.reporter.promgateway.factory.class: org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterFactory
  • 类型: push/基于 tags 格式
  • 参数:
    在这里插入图片描述
  • 配置示例:
metrics.reporter.promgateway.factory.class: org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterFactory
metrics.reporter.promgateway.hostUrl: http://localhost:9091
metrics.reporter.promgateway.jobName: myJob
metrics.reporter.promgateway.randomJobNameSuffix: true
metrics.reporter.promgateway.deleteOnShutdown: false
metrics.reporter.promgateway.groupingKey: k1=v1;k2=v2
metrics.reporter.promgateway.interval: 60 SECONDS

PrometheusPushGatewayReporter 发送器将运行指标发送给 Pushgateway,Prometheus 再从 Pushgateway 拉取、解析运行指标。

更多使用方法可查看 Prometheus 的文档

6)、StatsD

关于StatsD请参考官网信息。

  • factory.class
metrics.reporter.stsd.factory.class: org.apache.flink.metrics.statsd.StatsDReporterFactory
  • 类型: push/基于标志符格式

  • 参数:
    host - StatsD 的服务器地址。
    port - StatsD 的服务器端口。

  • 配置示例:

metrics.reporter.stsd.factory.class: org.apache.flink.metrics.statsd.StatsDReporterFactory
metrics.reporter.stsd.host: localhost
metrics.reporter.stsd.port: 8125
metrics.reporter.stsd.interval: 60 SECONDS

7)、Datadog

关于 Datadog官网了解更多的信息。

  • factory.class
metrics.reporter.dghttp.factory.class: org.apache.flink.metrics.datadog.DatadogHttpReporterFactory
  • 类型: push/基于 tags 格式
    1、使用 Datadog 时,Flink 运行指标中的任何变量,例如 、<job_name>、 <tm_id>、 <subtask_index>、<task_name>、 <operator_name>,都会被当作 host:localhost、job_name:myjobname 这样的 tag 发送。
    2、Note For legacy reasons the reporter uses both the metric identifier and tags. This redundancy can be avoided by enabling useLogicalIdentifier.

按照 Datadog 的 Histograms 命名约定,Histograms 类的运行指标会作为一系列 gauges 显示(<metric_name>.)。 默认情况下 min 即最小值被发送到 Datadog,sum 不会被发送。 与 Datadog 提供的 Histograms 相比,Histograms 类的运行指标不会按照指定的发送间隔进行聚合计算。

  • 参数:
    apikey - Datadog 的 API KEY。
    proxyHost - (可选的)发送到 Datadog 时使用的代理主机。
    proxyPort - (可选的)发送到 Datadog 时使用的代理端口,默认为 8080。
    dataCenter - (可选的)要连接的数据中心(EU/US),默认为 US。
    maxMetricsPerRequest - (可选的)每次请求携带的最大运行指标个数,默认为 2000。
    useLogicalIdentifier -> (optional) Whether the reporter uses a logical metric identifier, defaults to false.

  • 配置示例:

metrics.reporter.dghttp.factory.class: org.apache.flink.metrics.datadog.DatadogHttpReporterFactory
metrics.reporter.dghttp.apikey: xxx
metrics.reporter.dghttp.proxyHost: my.web.proxy.com
metrics.reporter.dghttp.proxyPort: 8080
metrics.reporter.dghttp.dataCenter: US
metrics.reporter.dghttp.maxMetricsPerRequest: 2000
metrics.reporter.dghttp.interval: 60 SECONDS
metrics.reporter.dghttp.useLogicalIdentifier: true

8)、Slf4j

本部分可以参考入门示例中的slf4j的例子。

  • factory.class
metrics.reporter.slf4j.factory.class: org.apache.flink.metrics.slf4j.Slf4jReporterFactory
  • 类型: push/基于标志符格式

  • 配置示例
    每分钟将指标写入到日志文件中
    flink-conf.yaml配置文件内容增加

metrics.reporter.slf4j.factory.class: org.apache.flink.metrics.slf4j.Slf4jReporterFactory
metrics.reporter.slf4j.interval: 60 SECONDS

以上,本文介绍了Flink的指标报告的类型以及几种发送器的示例配置,特别是针对jmx和slf4j给出了具体的实现示例及验证结果。由于本文是介绍flink的指标数据可以给哪些系统发送数据,至于具体实现算是各自的应用,故没有深入的展示。


http://www.niftyadmin.cn/n/5226817.html

相关文章

STM32g70开启定时器死机原因

在做低功耗产品时&#xff0c;检查发现由于之前开启了BOOTLOADER升级程序&#xff0c;修改了中断向量FALSH起始地址&#xff0c;只在KEIL TARGET IROM1中修改了&#xff0c; 而忘记在程序文件system_stm32f10x.c里修改中断向量表flash起始地址 system_stm32f10x.c里&#xff0…

spring cloud Eureka注册中心和Nacos注册中心

文章目录 Eureka注册中心.Eureka的结构和作用搭建eureka-server创建 服务引入eureka依赖编写启动类编写配置文件启动服务 服务注册1&#xff09;引入依赖2&#xff09;配置文件3&#xff09;启动多个user-service实例 服务发现1&#xff09;引入依赖2&#xff09;配置文件3&…

Linux-hid

/kernel/drivers/hid/hid-core.c hid总线驱动/kernel/drivers/hid/hid-generic.c hid通用驱动/kernel/drivers/hid/hid-multitouch.c 多点触控面板驱动/kernel/drivers/hid/hid-quirks.c hid-quirks.c 是 Linux 内核中的一个文件&#xff0c;用于实现 HID&#xff08;Human…

【Android知识笔记】架构专题(一)

什么是 MVC 其实我们日常开发中的Activity,Fragment和XML界面就相当于是一个MVC的架构模式,但往往Activity中需要处理绑定UI,用户交互,以及数据处理。 这种开发方式的缺点就是业务量复杂的时候一个Activity过于臃肿。但是页面结构不复杂的情况下使用这种方式就会显得很简…

Redis 入门和环境搭建

认识Redis Redis是一种NoSQL数据库&#xff0c;以键值对形式存储数据&#xff0c;支持多种数据结构&#xff0c;包括字符串、哈希、列表、集合、有序集合等&#xff0c;使其适用于多种应用场景。由于所有数据都存储在内存中&#xff0c;Redis的读写性能非常高。同时&#xff0…

第十七章 其他-rpc、rabbitmq(如何对消息做持久化、如何控制消息被消费的顺序)、celery(应用场景、运行机制、如何实现定时任务)

Python基础、函数、模块、面向对象、网络和并发编程、数据库和缓存、 前端、django、Flask、tornado、api、git、爬虫、算法和数据结构、Linux、设计题、其他 第十七章 其他 1. 什么是rpc&#xff1f; RPC&#xff08;Remote Procedure Call&#xff0c;远程过程调用&#x…

Python - Real-ESRGAN 提升图像、视频清晰度 - 最高可达 4 K

目录 一.引言 二.Real-ESRGAN 理论 1.模型简介 2.经典退化模型 ◆ 退化过程全览 ◆ K - 高斯滤波 ◆ N - 噪声 ◆ ↓r - Resize ◆ jpeg - 压缩 3.高阶退化模型 4.环形和超调伪影 5.网络结构 ◆ ESRGAN 生成器 ◆ U-Net 鉴别器 三.Real-ESRGAN 实战 1.快速体验…

【Linux 无网络状态下离线安装 MySQL】

【Linux 无网络状态下离线安装 MySQL】 一、准备安装包 可以去MySQL官网进行下载需要的安装包&#xff0c;下面这个链接 Mysql的下载地址 Microsoft Windows 版本 Linux-Generic 版本 5.7.4的版本和8.0版本的下载地址 全部版本的下载地址 libao的下载地址&#xff08;版…