Spring Boot+Java下Spark流处理 如何停止本地空目录的连续流式运行？

1. 无文件仍持续运行的本质原因 #

Spark 流式引擎（不管是旧DStream还是Structured Streaming）的默认设计是常驻监听模式：按照固定触发间隔扫描目标目录拉取新增数据，哪怕当前批次没有读取到任何文件，也不会自动退出，会持续循环执行空扫描任务。你日志里看到的每几毫秒就打印一次文件目录扫描结果、批次输入行数为0的情况，就是默认触发间隔过短（默认是尽可能快执行）导致的空跑。

2. 仅读取1个文件、剩余文件不处理的原因 #

• Spark 文件流源只认流式任务启动后新增、且完整原子写入的文件。如果你在流式任务启动前就把文件全部放入目标目录，任务首次扫描时会把这批文件的位置记录为初始偏移量，后续不会重复扫描读取这批存量文件；如果启动时目录已有1个文件被纳入首批处理，后续新增的2个文件就会被跳过
• 未配置checkpoint时，文件偏移量只存在内存中，容易因为状态丢失出现漏读、重复读的问题
• 边写边往监控目录放文件时，Spark会判定文件为不完整临时文件，直接跳过不处理

核心需求实现：无文件时自动停止流式任务 #

最适配你场景的方式是替换默认的连续触发模式，从“无限循环监听”改为“扫描全量可用文件、处理完即退出”，具体分两种场景：

场景1：不需要长期监听，处理完目录存量文件就退出 #

直接使用Trigger.AvailableNow()触发模式，替换原来的默认无限触发逻辑，修改后的processfileandconverttoexcel方法核心代码如下：

StructType schema = new StructType().add("filedata", DataTypes.StringType, true);
Dataset<Row> rawData = sessionconfig.readStream().format("csv").schema(schema)
        .csv(source);

StreamingQuery query1 = rawData.writeStream()
        .format("console")
        .outputMode(OutputMode.Update())
        // 触发规则：扫描当前所有未处理的可用文件，处理完成后自动终止查询，不会持续运行
        .trigger(Trigger.AvailableNow())
        // 持久化文件偏移量到checkpoint，避免漏读、重复读
        .option("checkpointLocation", "file:///c:/tmp/spark-checkpoint/file-converter")
        .start();

// 等待查询自动终止（无文件时扫描一次空批次就直接退出，有文件时处理完全部文件再退出）
query1.awaitTermination();
System.out.println("end of job ..............");

这种模式的效果：

• 目录无文件时：第一次扫描得到0输入行，执行完空批次就自动终止，不会持续循环
• 目录有3个文件时：会逐次识别所有未处理的文件，全部拉取到批次中处理，不会只读取1个
• 处理完成后直接退出流程，完全满足“无文件就停止连续运行”的需求

场景2：需要长期监听目录，但连续无数据时自动停止空跑 #

如果你需要保持流式监听能力，不想每次手动启动任务扫存量文件，可以增加自定义进度监听器，达到连续N个批次无数据就自动停止的效果：

// 配置10秒触发间隔，避免空跑频率太高耗资源
StreamingQuery query1 = rawData.writeStream()
        .format("console")
        .outputMode(OutputMode.Update())
        .trigger(Trigger.ProcessingTime(Durations.ofSeconds(10)))
        .option("checkpointLocation", "file:///c:/tmp/spark-checkpoint/file-converter")
        .start();

AtomicInteger emptyBatchCounter = new AtomicInteger(0);
// 注册流式查询进度监听器
query1.addListener(new StreamingQueryListener() {
    @Override
    public void onQueryStarted(QueryStartedEvent event) {}

    @Override
    public void onQueryProgress(QueryProgressEvent event) {
        long currentInputRows = event.progress().numInputRows();
        if (currentInputRows == 0) {
            int emptyCount = emptyBatchCounter.incrementAndGet();
            // 连续3个空批次（也就是30秒没有新文件）就自动停止查询
            if (emptyCount >= 3) {
                query1.stop();
                System.out.println("连续无新增数据，停止流式任务");
            }
        } else {
            // 检测到新数据时，重置空批次计数器
            emptyBatchCounter.set(0);
        }
    }

    @Override
    public void onQueryTerminated(QueryTerminatedEvent event) {}
});

query1.awaitTermination();

漏读文件的配套优化 #

• 不要在流式任务启动前就把文件放入监控目录，启动后等待第一次目录扫描完成，再把需要处理的文件放入目录，即可被全部识别
• 所有文件必须等完整写入（文件大小稳定、复制完成）后再放入监控目录，避免被判定为不完整临时文件跳过
• 生产环境部署时，把checkpointLocation配置为分布式持久化存储路径（比如HDFS），避免程序重启丢失文件偏移量状态

关于旧DStream API的说明 #

你test()方法中使用的JavaStreamingContext.textFileStream()是Spark 1.x时代的旧API，目前已经被Structured Streaming完全替代，存在维护性差、规则不透明的问题，后续生产环境部署建议直接替换为Structured Streaming实现，不需要单独维护两套流式逻辑。

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