监控指标

概览

我们为节点增加了指标计算。 用户添加 with 选项 inlong.metric.lables 后 Sort 会计算指标,inlong.metric.labels 选项的值由三部分构成:groupId={groupId}&streamId={streamId}&nodeId={nodeId}。 用户可以使用 metric reporter 去上报数据。

指标

支持的 extract 节点

支持节点级别指标

指标名Extract 节点描述
groupId_streamId_nodeId_numRecordsInkafka,mongodb-cdc,mysql-cdc,oracle-cdc,postgresql-cdc,pulsar,sqlserver-cdc输入记录数
groupId_streamId_nodeId_numBytesInkafka,mongodb-cdc,mysql-cdc,oracle-cdc,postgresql-cdc,pulsar,sqlserver-cdc输入字节数
groupId_streamId_nodeId_numRecordsInPerSecondkafka,mongodb-cdc,mysql-cdc,oracle-cdc,postgresql-cdc,pulsar,sqlserver-cdc每秒输入记录数
groupId_streamId_nodeId_numBytesInPerSecondkafka,mongodb-cdc,mysql-cdc,oracle-cdc,postgresql-cdc,pulsar,sqlserver-cdc每秒输入字节数

支持的 load 节点

支持节点级别指标

指标名Load 节点描述
groupId_streamId_nodeId_numRecordsOutclickhouse,elasticsearch,greenplum,hbase,
hdfs,hive,iceberg,kafka,
mysql,oracle,postgresql,sqlserver,tdsql-postgresql
输出记录数
groupId_streamId_nodeId_numBytesOutclickhouse,elasticsearch,greenplum,hbase,
hdfs,hive,iceberg,kafka,
mysql,oracle,postgresql,sqlserver,tdsql-postgresql
输出字节数
groupId_streamId_nodeId_numRecordsOutPerSecondclickhouse,elasticsearch,greenplum,
hbase,hdfs,hive,iceberg,
kafka,mysql,oracle,postgresql,sqlserver,tdsql-postgresql
每秒输出记录数
groupId_streamId_nodeId_numBytesOutPerSecondclickhouse,elasticsearch,greenplum,
hbase,hdfs,hive,iceberg,kafka,
mysql,oracle,postgresql,sqlserver,tdsql-postgresql
每秒输出字节数

用法

这里将介绍一个同步MYSQL数据到PostgreSQL的例子,同时介绍指标的使用。

  • flink sql 的使用
  1. create table `table_groupId_streamId_nodeId1`(
  2. `id` INT,
  3. `name` INT,
  4. `age` STRING,
  5. PRIMARY KEY(`id`) NOT ENFORCED)
  6. WITH (
  7. 'connector' = 'mysql-cdc-inlong',
  8. 'hostname' = 'xxxx',
  9. 'username' = 'xxx',
  10. 'password' = 'xxx',
  11. 'database-name' = 'test',
  12. 'scan.incremental.snapshot.enabled' = 'true',
  13. 'server-time-zone' = 'GMT+8',
  14. 'table-name' = 'user',
  15. 'inlong.metric' = 'mysqlGroup&mysqlStream&mysqlNode1'
  16. );
  17. CREATE TABLE `table_groupId_streamId_nodeId2`(
  18. PRIMARY KEY (`id`) NOT ENFORCED,
  19. `id` INT,
  20. `name` STRING,
  21. `age` INT)
  22. WITH (
  23. 'connector' = 'jdbc-inlong',
  24. 'url' = 'jdbc:postgresql://ip:5432/postgres',
  25. 'username' = 'postgres',
  26. 'password' = 'inlong',
  27. 'table-name' = 'public.user',
  28. 'inlong.metric' = 'pggroup&pgStream&pgNode'
  29. );
  30. INSERT INTO `table_groupId_streamId_nodeId2`
  31. SELECT
  32. `id`,
  33. `name`,
  34. `age`
  35. FROM `table_groupId_streamId_nodeId1`;
  • 我们可以在flink-conf.yaml中添加metric report配置
  1. metric.reporters: promgateway
  2. metrics.reporter.promgateway.class: org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporter
  3. metrics.reporter.promgateway.host: ip
  4. metrics.reporter.promgateway.port: 9091
  5. metrics.reporter.promgateway.interval: 60 SECONDS

ipport 是你的 pushgateway 的配置。

  • 执行上面的sql后,我们可以访问 pushgateway 的 url: http://ip:port

当我们使用的 metric report 是 org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporter 指标名将添加前缀 flink_taskmanager_job_task_operator
我们可以看到完整的指标名如下:
flink_taskmanager_job_task_operator_groupId_streamId_nodeId_numRecordsIn,
flink_taskmanager_job_task_operator_groupId_streamId_nodeId_numBytesIn,
flink_taskmanager_job_task_operator_groupId_streamId_nodeId_numRecordsInPerSecond,
flink_taskmanager_job_task_operator_groupId_streamId_nodeId_numBytesInPerSecond,
flink_taskmanager_job_task_operator_groupId_streamId_nodeId_numRecordsOut,
flink_taskmanager_job_task_operator_groupId_streamId_nodeId_numBytesOut,
flink_taskmanager_job_task_operator_groupId_streamId_nodeId_numRecordsOutPerSecond,
flink_taskmanager_job_task_operator_groupId_streamId_nodeId_numBytesOutPerSecond.