如何在Spark结构化流中获取写入记录数？

Question

3

我有几个结构化流配置在一个Spark会话中。我需要知道每个流中读取和写入了多少记录。

例如，我有以下两个流：

1. read-s3 -> transform -> write-s3 2. read-s3 -> transform -> write-db

我知道可以使用SparkListener().onTaskEnd()来获取信息，但此时我没有查询名称，而taskEnd.taskMetrics().outputMetrics().recordsWritten()始终为0，因此这不是一个选项。

另一种方法是在dataset.map()中使用累加器进行增量计算。但这只能得到要写入的记录数（如果sink未失败）。

除此之外，我尝试使用StreamingQueryListener（我用它来获取numInputRows），但我找不到任何关于写入记录数量的指标。

是否有可能获取此类指标？

- Yuriy Bondaruk

这还是关于自定义sink的问题吗？ - Alper t. Turker

不，那篇帖子是关于修复若干输入行的问题，而我在这里问的是输出记录的数量，无论使用哪种接收器都没关系。当然，如果我使用自定义接收器，我可以自己添加这些指标，但如果我使用FileStreamSink该怎么办呢？ - Yuriy Bondaruk

FileStreamSink中存在一个bug（https://issues.apache.org/jira/browse/SPARK-23288），已在2.3.1版本中修复。 - Yuriy Bondaruk

请将以下与编程相关的内容从英文翻译成中文。只返回翻译后的文本：请将其作为答案发布（https://stackoverflow.com/help/self-answer）。这样更容易被未来的访问者看到，并且可以解决问题。 - Alper t. Turker

1个回答

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- Yuriy Bondaruk · Answer 1

作为一个解决方法，在写入到 sink 之前，在映射函数中使用累加器来计算记录数。