topic Re: maxFilesPerTrigger not working while loading data from Unity Catalogue table in Get Started Discussions

maxFilesPerTrigger not working while loading data from Unity Catalogue table

sanjay — Wed, 28 Aug 2024 12:08:36 GMT

Hi,

I am using streaming on unity catalogue tables and trying to limit the number of records read in each batch. Here is my code but its not respecting maxFilesPerTrigger, instead reads all available data.

(spark.readStream

.option("skipChangeCommits","true")

.option("useNotifications","true")

.option("includeExistingFiles","true")

.option("allowOverwrites",True)

.option("ignoreMissingFiles",True)

.option("maxFilesPerTrigger", 2)

.table(table_path)

.writeStream

.option("checkpointLocation", checkpoint_path)

.trigger(availableNow=True)

.foreachBatch(foreachBatchFunction)

.start()

.awaitTermination())

Appreciate any help here.

Re: maxFilesPerTrigger not working while loading data from Unity Catalogue table

Witold — Wed, 28 Aug 2024 13:03:50 GMT

How does your table structure look like, can you confirm that all your data is actually in more than just two files?

Re: maxFilesPerTrigger not working while loading data from Unity Catalogue table

sanjay — Wed, 28 Aug 2024 17:02:15 GMT

My table has multiple rows e.g. lets take simple table as employee with emd_id & emp_name columns and use streaming to process any updates to this table. In case there are more than 2 inserts, i want to process max 2 rows at a time

Re: maxFilesPerTrigger not working while loading data from Unity Catalogue table

sanjay — Thu, 29 Aug 2024 07:31:48 GMT

am able to resolve the issue. not sure what was the issue. Its working now without any code change.

Re: maxFilesPerTrigger not working while loading data from Unity Catalogue table

Witold — Thu, 29 Aug 2024 07:36:54 GMT

I believe you misunderstand the fundamentals of delta tables. `maxFilesPerTrigger` has nothing to do with how many rows you will process at the same time. And if you really want to control the number of records per file, then you need to adapt the writer accordingly. Besides that fact that having two records per file is a very bad idea for multiple reasons. Depending on your load, you will end up with thousand of small files and dozens of useless history entries in the delta table log.

Of course I don't know your use case behind it, but I'm pretty sure that you don't want to process two rows at the same. Keep in mind that you're in a big data environment, and it's designed to process millions of rows at the same time.

Re: maxFilesPerTrigger not working while loading data from Unity Catalogue table

sanjay — Thu, 29 Aug 2024 07:46:57 GMT

Thank you Witold, 2 was just an example. I am having thousands of files coming every second and want to limit files per batch otherwise process gets struct if there are too many files in given batch.

I am able to limit the batch size while running single job but not as continuous job.

(spark.readStream

.option("skipChangeCommits","true")

.option("useNotifications","true")

.option("includeExistingFiles","true")

.option("allowOverwrites",True)

.option("ignoreMissingFiles",True)

.option("maxFilesPerTrigger", 2)

.table(table_path)

.writeStream

.option("checkpointLocation", checkpoint_path)

.trigger(processingTime="1 second")

.foreachBatch(foreachBatchFunction)

.start())