在Spark结构化流中比较“numInputRows”和“numOutputRows”的最佳方法是什么?

我在日志中得到这一点,但我如何将这些指标推到一些数据库,如石墨,这样我就可以在每个MicroBatchExecution比较“numInputRows”和“numOutputRows”的值。

 INFO MicroBatchExecution: Streaming query made progress: {
  "id" : "34ad5227-6f3d-436c-a278-c1aa67b70aaf",
  "runId" : "86143685-e702-48b2-8b48-35b16d092de5",
  "name" : null,
  "timestamp" : "2021-09-12T06:46:29.879Z",
  "batchId" : 596,
  "numInputRows" : 180,
  "inputRowsPerSecond" : 43.53083434099154,
  "processedRowsPerSecond" : 48.27031375703942,
  "durationMs" : {
    "addBatch" : 3399,
    "getBatch" : 0,
    "latestOffset" : 234,
    "queryPlanning" : 7,
    "triggerExecution" : 3729,
    "walCommit" : 50
  },
  "stateOperators" : [ ],
  "sources" : [ {
    "description" : "KafkaV2[Subscribe[topic1, topic2]]",
    "startOffset" : {
      "topic2" : {
        "0" : 21
      },
      "topic1" : {
        "0" : 1365446173
      }
    },
    "endOffset" : {
      "topic2" : {
        "0" : 21
      },
      "topic1" : {
        "0" : 1365446353
      }
    },
    "numInputRows" : 180,
    "inputRowsPerSecond" : 43.53083434099154,
    "processedRowsPerSecond" : 48.27031375703942
  } ],
  "sink" : {
    "description" : "FileSink[/tmp/local_kafka_graphite/output/]",
    "numOutputRows" : -1
  }
}
阅读全文

▼ 版权说明

相关文章也很精彩
推荐内容
更多标签
相关热门
全站排行
随便看看

错说 cuoshuo.com —— 程序员的报错记录

部分内容根据CC版权协议转载;网站内容仅供参考,生产环境使用务必查阅官方文档

辽ICP备19011660号-5

×

扫码关注公众号:职场神器
发送: 1
获取永久解锁本站全部文章的验证码