How to limit number of files in each batch in streaming batch processing

[sanjay]

Hi,

I am running batch job which processes incoming files. I am trying to limit number of files in each batch process so added maxFilesPerTrigger option. But its not working. It processes all incoming files at once.

(spark.readStream.format("delta").load(silver_path)

.writeStream

.option("checkpointLocation", gold_checkpoint_path)

.option("maxFilesPerTrigger", 200)

.trigger(once=True)

.foreachBatch(foreachBatchFunction)

.start()

.awaitTermination()

)

Please suggest.

Regards,

Sanjay

[-werners-]

can you try with trigger = availablenow

[sanjay]

Tried available now, but its also processing all data available for processing. I want to process in batch, max 200 files i each batch though I have 1,000 files to process.

[-werners-]

ok, how do you know that 1000 files are selected?

I ask because delta lake (your source) also stores old versions of data, which will not be sent to the stream. Physically your delta lake might have 1000 files but the current state is maybe only 150 files -> 1 microbatch.

Is that possible?

[sanjay]

I have send 1000 files to process in previous layer and I don't want to process all in one go. I can see all 1000 received in current batch

[-werners-]

I think I found the issue.

The maxfilespertrigger option has to be set on the source, not on the sink (as you do).

Try to move the option before the load statement..

so readstream.option().load()...

[sanjay]

Still getting all 1000 files.

(spark.readStream.format("delta").option("maxFilesPerTrigger", 100).load(silver_path)

.writeStream

.option("checkpointLocation", gold_checkpoint_path)

.trigger(once=True)

.foreachBatch(foreachBatchFunction)

.start()

.awaitTermination()

)

[-werners-]

spark.readStream.format("delta")

.option("maxFilesPerTrigger", "100")

.load(<table>)

.writeStream

.format("delta")

.outputMode("append")

.option("checkpointLocation", "...")

.table(<table2>)

[sanjay]

Sorry not an expert in this. But how to process my custom code..

.foreachBatch(foreachBatchFunction)

.start()

.awaitTermination()

[-werners-]

Sorry, it is basically only the part up to load() that is important.

Also try to enter the number of files as a string instead of an int.

[sanjay]

Still getting full load.

df = (spark.readStream.format("delta")

.option("maxFilesPerTrigger", "100")

.load(silver_path)

)

(df.writeStream

.option("checkpointLocation", gold_checkpoint_path)

.trigger(once=True)

.foreachBatch(foreachBatchFunction)

.start()

.awaitTermination())

[-werners-]

Can you try to also set the maxFilespertrigger in the sink?

spark.readStream.format("delta")

.option("maxFilesPerTrigger", "100")

.load(silver_path)

.writeStream

.option("checkpointLocation", gold_checkpoint_path)

.option("maxFilesPerTrigger", "100")

.trigger(once=True)

.foreachBatch(foreachBatchFunction)

.start()

[-werners-]

here is also a SO topic on how you can test it:

https://stackoverflow.com/questions/70134468/spark-structured-streaming-rate-limit

[sanjay]

no, still getting all 1000 files

[-werners-]

strange, it should work.