count() 可以在 Spark 中被调用两次吗？

Question

我使用的是结构化流从redis加载记录，代码如下

val image = spark.readStream.format("redis").schema(...).load()
val q = image.writeStream.foreachBatch { 
  (batchDF: DataFrame, batchId: Long) => {
    logger.info(batchDF.count()) // INFO: 4 (or other number)
    logger.info(batchDF.count()) // INFO: 0
    logger.info(batchDF.count()) // INFO: 0
  }
}.start()
q.awaitTermination()

如上所示，第一个 logger.info 似乎得到了 batchDF 的正确计数，但从第二个开始它是 0。

更重要的是，第一个 count() 之后 batchDF 消失了吗？

Answer 1

So after count() is called once, is the RDD still there?

-- 是的，您可以 运行 对 RDD 执行多项操作。它退出直到 spark session alive。

检查计数是否为零。您可以使用如下 rdd.isEmpty 函数：

if(rdd.isEmpty)
{// means count is zero
} else {//means not zero}