Spark Streaming无新数据仍执行foreachRDD？如何仅在有新文件时触发？

你遇到的这个情况其实是Spark 1.6版本中textFileStream的默认行为：它会按照你设置的批次间隔（这里是2秒）定期扫描目标目录，不管目录里有没有新增文件，每个批次都会生成一个对应的RDD——如果目录为空或者没有新文件，这个RDD就是空RDD，所以foreachRDD会被触发执行，打印出0。

要实现「仅当文件夹中有新文件时才执行逻辑」，你只需要在foreachRDD中先判断当前RDD是否非空，再执行后续代码即可。

修改后的示例代码 #

val conf = new SparkConf().setAppName("Name").setMaster("local")
val sc = new SparkContext(conf)
val ssc = new StreamingContext(sc, Seconds(2))
val directoryStream = ssc.textFileStream("""file:///C:/Users/something/something""")

directoryStream.foreachRDD(file => {
  // 先判断RDD是否包含数据
  if (!file.isEmpty()) {
    println(file.count())
    // 这里添加你的业务逻辑
  }
})

ssc.start()
ssc.awaitTermination()

补充说明 #

• file.isEmpty()是一个Spark Action操作，它会快速检查RDD是否有数据（不会全量扫描数据，性能开销很小），只有当RDD非空时才会进入代码块执行你的逻辑。
• 在Spark 1.6中，textFileStream没有内置的「仅在有新数据时触发批次」的配置，所以这种手动判断的方式是最直接有效的解决方案。
• 注意：如果你后续有更复杂的业务逻辑，确保所有需要在有数据时执行的代码都放在if块内部，避免空批次带来的无效执行。

内容的提问来源于stack exchange，提问作者osk