如何限製在每一批數量的文件流批處理

[桑傑]

你好,

我運行批處理作業流程的文件。我試圖限製在每個批處理文件數量所以添加maxFilesPerTrigger選項。但它不工作。它處理所有輸入文件。

(spark.readStream.format(“δ”).load (silver_path)

.writeStream

gold_checkpoint_path .option (“checkpointLocation”)

.option (“maxFilesPerTrigger”, 200年)

.trigger(一旦= True)

.foreachBatch (foreachBatchFunction)

.start ()

.awaitTermination ()

)

請建議。

問候,

桑傑

[werners1]

你可以嚐試與觸發= availablenow

[桑傑]

現在嚐試可用,但它還處理所有數據進行處理。我想在批處理過程,最大200個文件我每一批雖然我有1000個文件的過程。

[werners1]

好吧,你怎麼知道1000個文件被選中?

我問,因為三角洲湖(源代碼)也將舊版本的數據存儲,這將不會被發送到流中。身體三角洲湖可能有1000個文件但當前狀態也許隻有150個文件- > 1 microbatch。

這有可能嗎?

[桑傑]

我已經發送1000個文件處理前層和我不想處理所有。我可以看到所有1000收到了在當前批