我有一个应用程序，它从Kafka获取数据并将其保存到数据库。我的代码如下所示：

spark.readStream
  .format("kafka")
  .options(options)
  .load()
  .writeStream
  .trigger(Trigger.ProcessingTime(20000))
  .foreachBatch({ (batch: DataFrame, _: Long) =>
    val rowsCount = batch.count
    saveBatch(batch)
    println(s"Saved $rowsCount rows")
  })
  .start()

字符串
在Spark UI中，我查看Structure Streaming选项卡，看到我的流的处理速率是每秒10万行。如果我像这样删除行数：

.foreachBatch({ (batch: DataFrame, _: Long) =>
    saveBatch(batch)
  })
  .start()

型
如你所见，在第一种情况下，我没有使用缓存，当我计算批处理中的行数并将批处理保存到数据库时，我可能从Kafka读取数据两次（而不是第二种情况，当阅读只有一次）我不能相信，这样的琐碎操作，如行计数导致额外的阅读从Kafka，它是任何方法，在不缓存数据的情况下获取批处理长度？