即使目录中没有文件或数据，Spark 也会继续从本地目录流式传输

Question

我的代码遇到了挑战。即使我的本地目录中没有数据或文件，spark 也会继续流式传输。尽管我仍将部署到生产环境，但我正在本地系统上运行 spark。无论如何我都是 spark 新手。即使我添加了三个文件，spark 也只会选择一个，并且不会流式传输其余文件，但我担心的是连续流式传输。请问在没有文件或数据的情况下如何控制并停止这种连续流式传输？我正在使用带有 java 的 spring boot。以下是我的源代码：

@Component
public class FileConverter {

    @Value("${output.path}")
    protected String source;
    @Autowired
    SparkSessionConfig session;

    public void processfileandconverttoexcel() throws TimeoutException, StreamingQueryException {
        StreamingQuery query = null;
        SparkSession sessionconfig = session.getSession();
        System.out.println(sessionconfig.version());
        // define schema type of file data source
        sessionconfig.sparkContext().setLogLevel("ERROR");
        StructType schema = new StructType().add("filedata", DataTypes.StringType, true);
        // build the streaming data reader from the file source, specifying csv file
        Dataset<Row> rawData = sessionconfig.readStream().format("csv").schema(schema)
                .csv(source);
        rawData.createOrReplaceTempView("filedata");
        Dataset<Row> results = sessionconfig.sql("select * from filedata");
        StreamingQuery query1 = rawData.writeStream().format("console").outputMode(OutputMode.Update()).start();
        query1.awaitTermination();
        System.out.println("end ofjob ..............");
    }
      *this test method when tried and called for streaming, is not even reading any files but continues to stream*
    public void test() throws InterruptedException {
        SparkConf conf = new SparkConf().setMaster("local[5]").setAppName("NetworkWordCount");
        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(1));
        JavaDStream<String> lines = jssc.textFileStream(source);
        lines.print();
        lines.foreachRDD((rdd, time) -> {
            ;
            if (!rdd.isEmpty()) {
                lines.print();
                rdd.foreach(f -> {
                    System.out.println(f);
                });
            }
        });
        jssc.start(); // Start the computation
        jssc.awaitTermination(); // Wait for the computation to terminate
    }
}
@Component
public class SparkSessionConfig {
   
    public SparkSession getSession() {
        System.out.println("creating spark session ,,,,,,,,,,,,,,, ");
         System.setProperty("hadoop.home.dir", "C:/winutils");
         SparkSession spark  = SparkSession.builder().appName("File Converter")
                 .config("spark.sql.dir.warehouse.dir","file:///c:/tmp/")
                 .master("local[5]")
                 .getOrCreate()  ;
         System.out.println("spark session created ,,,,,,,,,,,,,,, ");
        return spark;
        
    }
}

这是我的控制台输出总结

2021-01-13 03:09:42.711  INFO 12784 --- [4-7d949befed27]] o.a.s.s.e.datasources.InMemoryFileIndex  : It took 1 ms to list leaf files for 1 paths.
2021-01-13 03:09:42.723  INFO 12784 --- [4-7d949befed27]] o.a.s.s.e.datasources.InMemoryFileIndex  : It took 1 ms to list leaf files for 1 paths.
2021-01-13 03:09:42.734  INFO 12784 --- [4-7d949befed27]] o.a.s.s.e.datasources.InMemoryFileIndex  : It took 1 ms to list leaf files for 1 paths.
2021-01-13 03:09:42.736  INFO 12784 --- [4-7d949befed27]] o.a.s.s.e.streaming.MicroBatchExecution  : Streaming query made progress: {
  "id" : "2445367b-fc80-416e-9ff5-0cddd1af7bb2",
  "runId" : "1dc94f26-0bde-4447-b204-7d949befed27",
  "name" : null,
  "timestamp" : "2021-01-13T11:09:42.733Z",
  "batchId" : 1,
  "numInputRows" : 0,
  "inputRowsPerSecond" : 0.0,
  "processedRowsPerSecond" : 0.0,
  "durationMs" : {
    "latestOffset" : 2,
    "triggerExecution" : 2
  },
  "stateOperators" : [ ],
  "sources" : [ {
    "description" : "FileStreamSource[file:/C:/Users/DELL/Desktop/sourcefolders]",
    "startOffset" : {
      "logOffset" : 0
    },
    "endOffset" : {
      "logOffset" : 0
    },
    "numInputRows" : 0,
    "inputRowsPerSecond" : 0.0,
    "processedRowsPerSecond" : 0.0
  } ],
  "sink" : {
    "description" : "org.apache.spark.sql.execution.streaming.ConsoleTable$@2ab718be",
    "numOutputRows" : 0
  }
}
2021-01-13 03:09:42.748  INFO 12784 --- [4-7d949befed27]] o.a.s.s.e.datasources.InMemoryFileIndex  : It took 1 ms to list leaf files for 1 paths.
2021-01-13 03:09:42.761  INFO 12784 --- [4-7d949befed27]] o.a.s.s.e.datasources.InMemoryFileIndex  : It took 1 ms to list leaf files for 1 paths.
2021-01-13 03:09:42.773  INFO 12784 --- [4-7d949befed27]] o.a.s.s.e.datasources.InMemoryFileIndex  : It took 1 ms to list leaf files for 1 paths.

请大家帮忙看看。

Answer 1

导入Trigger类后，尝试将触发方法设为Once

import org.apache.spark.sql.streaming.Trigger

StreamingQuery query1 = rawData.writeStream().format("console").trigger(Trigger.Once()).outputMode(OutputMode.Update()).start();
query1.awaitTermination();