Autoloader maxFilesPerTrigger not working correctly

[sanjay]

Hi,

am trying to apply batch size in autoloader and code is as below. But its picking all the changes in one go even if I have put maxFilesPerTrigger as 10. Appreciate any help.

(spark.readStream.format("json").schema(streamSchema)

.option("cloudFiles.backfillInterval", "1 day")

.option("cloudFiles.fetchParallelism", 100)

.option("cloudFiles.useNotification","true")

.option("cloudFiles.includeExistingFiles","true")

.option("cloudFiles.allowOverwrites",True)

.option("ignoreMissingFiles",True)

.option("maxFilesPerTrigger", 10)

.load(raw_path)

.writeStream

.option("checkpointLocation", bronze_checkpoint_path)

.trigger(once=True)

.foreachBatch(foreachBatchFunction)

.start()

.awaitTermination()

)

[Lakshay · Accepted solution]

Hi @Sanjay Jain​ , Since you have provided the trigger as once, the maxFilesPerTrigger will not take effect here. With trigger once, all the files will be read together. You need to change the trigger for this option to come into effect.

Please refer the document here :

https://docs.databricks.com/ingestion/auto-loader/options.html#common-auto-loader-options

[sanjay]

Thank you Lakshay

[sanjay]

Hi,

This is working for autoloader and I am able to limit batch size. But in next layer, I am still getting full load. Is there any way to limit batch size. Here is my code.

(spark.readStream.format("delta")

.option("cloudFiles.useNotification","true")

.option("cloudFiles.includeExistingFiles","true")

.option("cloudFiles.allowOverwrites",True)

.option("ignoreMissingFiles",True)

.option("cloudFiles.maxFilesPerTrigger", 100)

.option("ignoreChanges","true")

.load(bronze_path)

.writeStream

.option("checkpointLocation", silver_checkpoint_path)

.trigger(processingTime="1 minute")

.foreachBatch(foreachBatchFunction)

.start()

)