Проблема с потоковой передачей файлов Spark

API textFileStream не предназначен для чтения существующего содержимого каталога, вместо этого он предназначен для отслеживания заданного пути файловой системы, совместимого с Hadoop, на наличие изменений, файлы должны быть записаны в отслеживаемое местоположение путем «перемещения» их из другого расположение в той же файловой системе. Короче говоря, вы подписываетесь на изменения каталога и будете получать содержимое вновь появившихся файлов в отслеживаемом расположении — в том состоянии, в котором файл(ы) появляются на момент снимка мониторинга (то есть 2000 мс в вашем случае), и любые дальнейшие обновления файлов не будут достигать потока, будут выполняться только обновления каталога (новые файлы).

Вы можете эмулировать обновления, создавая новый файл во время сеанса мониторинга:

import org.apache.commons.io.FileUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;

import java.io.File;
import java.io.IOException;
import java.util.List;

public class DStreamExample {

public static void main(final String[] args) throws IOException {

    final SparkConf sparkConf = new SparkConf();
    sparkConf.setAppName("SparkJob");
    sparkConf.setMaster("local[4]"); // for local

    final JavaSparkContext sc = new JavaSparkContext(sparkConf);

    final JavaStreamingContext ssc = new JavaStreamingContext(sc,
            new Duration(2000));

    final JavaDStream<String> lines = ssc.textFileStream("/opt/test/");

    // spawn the thread which will create new file within the monitored directory soon
    Runnable r = () -> {
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        try {
            FileUtils.write(new File("/opt/test/newfile1"), "whatever");
        } catch (IOException e) {
            e.printStackTrace();
        }
    };

    new Thread(r).start();


    lines.foreachRDD((Function<JavaRDD<String>, Void>) rdd -> {
        List<String> lines1 = rdd.collect();
        lines1.stream().forEach(l -> System.out.println(l));
        return null;
    });

    ssc.start();
    ssc.awaitTermination();
}

}