如何计算从 Kafka 获取记录的时间？

Question

我有一个简单的工作，触发器=15 秒，Source=Kafka 和 Sink=S3。是否有可能找到从 Kafka 下载消息需要多少时间？或者说如果我有 Sink=Console，它带回驱动上的数据，是否有可能找到从 Kafka 下载数据的时间以及将其带回驱动的时间？

在写入 S3 时，我从驱动程序中获取这些用于查询。是否有可能在 triggerExecution = 44 秒内从 Kafka 下载 99998 行花费了多少时间？

Streaming query made progress: {
  id : 1383g52b-8de4-4e95-a3s9-aea73qe3ea56,
  runId : 1206f5tc-t503-44r0-bc0c-26ce404w6724,
  name : null,
  timestamp : 2017-08-25T01:42:10.000Z,
  numInputRows : 99998,
  inputRowsPerSecond : 1666.6333333333334,
  processedRowsPerSecond : 2263.9860535669814,
  durationMs : {
    addBatch : 42845,
    getBatch : 3,
    getOffset : 68,
    queryPlanning : 6,
    triggerExecution : 44169,
    walCommit : 1245
  },
  stateOperators : [ ],
  sources : [ {
    description : KafkaSource[Subscribe[kafka_topic]],
    startOffset : {
      kafka_topic : {
        2 : 20119244,
        4 : 20123550,
        1 : 20124601,
        3 : 20113622,
        0 : 20114208
      }
    },
    endOffset : {
      kafka_topic : {
        2 : 20139245,
        4 : 20143531,
        1 : 20144592,
        3 : 20133663,
        0 : 20134192
      }
    },
    numInputRows : 99998,
    inputRowsPerSecond : 1666.6333333333334,
    processedRowsPerSecond : 2263.9860535669814
  } ],
  sink : {
    description : FileSink[s3://s3bucket]
  }
}

谢谢！

Answer 1

您应该通过查看StreamingQuery.lastProgress.durationMs 找到问题的答案。

按照他们的计算顺序，以下持续时间告诉您：

• getOffset是获取所有来源的偏移量的时间

• getBatch 是从所有来源（一个接一个，按顺序）获取流数据集（又名批次）的时间。

• addBatch 是将流数据集写入接收器的时间

话虽如此......

是否可以计算从 Kafka 下载消息需要多少时间？

这是addBatch 持续时间（因为那是数据集在执行器上作为 RDD 执行的时间）

是否可以了解在 triggerExecution = 44 秒内从 Kafka 下载 99998 行花费了多少时间？

您必须将 addBatch 持续时间与 StreamingQuery.recentProgress 数组相加。

Answer 2

由于从 Kafka 读取和读取记录的处理是流水线的，因此很难找到读取的确切时间。

而且很多时候这并不重要，因为处理是瓶颈，而不是从 Kafka 读取。所以真正的问题是，你为什么要关心确切的 Kafka 读取时间？