Re: Autoloader maxFilesPerTrigger not working corr...

sanjay · ‎06-05-2023

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 · ‎06-07-2023

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

View solution in original post

sanjay · ‎06-07-2023

Thank you Lakshay

sanjay · ‎06-09-2023

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()

)

Autoloader maxFilesPerTrigger not working correctly