我使用spark结构化流媒体读取Kafka主题的记录；我打算统计spark中每个“微批次”收到的记录数 readstream 这是一个片段：

val kafka_df = sparkSession
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host:port")
  .option("subscribe", "test-count")
  .load()

我从文件中了解到，Kafka将被懒惰地评估，当一个 streamingQuery 开始（下一步），当它被评估时，它持有一个微批。所以，我想 groupBy 主题后接 count 应该有用。
这样地：

val counter = kafka_df
             .groupBy("topic")
             .count()

现在要评估所有这些，我们需要一个streaminquery，比方说，一个console sink查询来在控制台上打印它。这就是问题所在。一个关于 aggregate Dataframe，例如 kafka_df 仅适用于 outputMode 完成/更新而不是附加。
这实际上意味着streamingquery报告的计数是累积的。
这样地：

val counter_json = counter.toJSON   //to jsonify 
 val count_query = counter_json
                   .writeStream.outputMode("update")
                   .format("console")
                   .start()          // kicks of lazy evaluation
                   .awaitTermination()

在受控设置中，其中：
实际发表记录：1500
实际接收微批次：3
实际接收记录：1500
每个微批的计数应该是500，所以我希望（希望）查询打印到控制台：
主题：测试计数
计数：500
主题：测试计数
计数：500
主题：测试计数
计数：500
但事实并非如此。它实际上打印了：
主题：测试计数
计数：500
主题：测试计数
count:1000
主题：测试计数
计数：1500
我理解这是因为“outputmode”完成/更新（累计）
我的问题是：有没有可能准确地得到每一个微批量是spark kafka结构化流媒体的计数？
从文档中，我发现了水印方法（支持append）：

val windowedCounts = kafka_df
                    .withWatermark("timestamp", "10 seconds")
                    .groupBy(window($"timestamp", "10 seconds", "10       seconds"), $"topic")
                    .count()

 val console_query = windowedCounts
                    .writeStream
                    .outputMode("append")
                    .format("console")
                    .start()
                    .awaitTermination()

但结果是 console_query 是不准确的，而且看起来很离谱。
热释光；博士-任何关于准确计算spark kafka微批次记录的想法都将不胜感激。