如何获取Spark Streaming处理的记录总数？

Question

有谁知道 Spark 如何计算其记录数（我认为它与批次中的事件数相同），如此处所示？

我试图弄清楚如何远程获取该值（UI 中的流选项不存在 REST-API）。

基本上我正在尝试做的是获取我的应用程序处理的记录总数。我需要此信息用于门户网站。

我尝试计算Records每个阶段的数量，但它给了我完全不同的数字，如上图所示。每个阶段都包含有关其记录的信息。如图所示

我正在使用这个简短的 python 脚本来计算每个阶段的“inputRecords”。这是源代码：

import json, requests, urllib
print "Get stages script started!"
#URL REST-API
url = 'http://10.16.31.211:4040/api/v1/applications/app-20161104125052-0052/stages/'
response = urllib.urlopen(url)
data = json.loads(response.read())

stages = []
print len(data)
inputCounter = 0
for item in data:
        stages.append(item["stageId"])
        inputCounter += item["inputRecords"]
print "Records processed: " + str(inputCounter)

如果我理解正确的话：每个Batch都有一个Job，每个Job都有多个Stages，这些Stages都有多个Tasks。

所以对我来说，计算每个 的输入是有意义的Stage。

Answer 1

Spark 在驱动程序上提供了一个指标端点：

<driver-host>:<ui-port>/metrics/json

Spark Streaming 应用程序将报告 UI 中可用的所有指标以及更多指标。您可能正在寻找的是：

<driver-id>.driver.<job-id>.StreamingMetrics.streaming.totalProcessedRecords: {
value: 48574640
},
<driver-id>.driver.<job-id>.StreamingMetrics.streaming.totalReceivedRecords: {
value: 48574640
}

该端点可以定制。有关信息，请参阅Spark 指标。