topic Re: How to limit number of files in each batch in streaming batch processing in Data Engineering

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

sanjay — Thu, 30 Mar 2023 06:59:29 GMT

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

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

-werners- — Thu, 30 Mar 2023 08:53:56 GMT

can you try with trigger = availablenow

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

sanjay — Thu, 30 Mar 2023 09:20:20 GMT

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.

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

-werners- — Thu, 30 Mar 2023 09:36:27 GMT

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?

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

sanjay — Thu, 30 Mar 2023 10:00:59 GMT

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

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

-werners- — Thu, 30 Mar 2023 10:15:38 GMT

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

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

sanjay — Thu, 30 Mar 2023 10:22:28 GMT

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

)

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

-werners- — Thu, 30 Mar 2023 10:29:16 GMT

spark.readStream.format("delta")

.option("maxFilesPerTrigger", "100")

.load(<table>)

.writeStream

.format("delta")

.outputMode("append")

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

.table(<table2>)

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

sanjay — Thu, 30 Mar 2023 10:33:19 GMT

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

.foreachBatch(foreachBatchFunction)

.start()

.awaitTermination()

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

-werners- — Thu, 30 Mar 2023 10:35:09 GMT

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.

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

sanjay — Thu, 30 Mar 2023 10:46:09 GMT

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

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

-werners- — Thu, 30 Mar 2023 10:49:45 GMT

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

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

-werners- — Thu, 30 Mar 2023 10:51:39 GMT

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

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

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

sanjay — Thu, 30 Mar 2023 10:52:22 GMT

no, still getting all 1000 files

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

-werners- — Thu, 30 Mar 2023 10:57:55 GMT

strange, it should work.

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

Sandeep — Thu, 30 Mar 2023 14:02:02 GMT

@Sanjay Jain , inside your gold_checkpoint_path, there are a few subfolders.

Go to "commits" and check which is the latest file inside (You can see files named 1,2,3,4,.....50,51 so on. File named with the highest number is the latest one. Assume it is 60 for example. This means micro batch 60 is committed. If no batch is committed yet, you will see no files).

And then check for files inside "offsets" folder. see the latest one in that folder too. That will in almost all cases you will see a file with name = latest batchID found in commits + 1 (61 as per this example. If there were no files at all inside commits, then you will see a file named "0" inside this folder.). And If you see this behavior, take a backup of this latest file and then delete it. Then restart the job. This should help!

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

sanjay — Thu, 30 Mar 2023 14:51:33 GMT

This seems to be manual step, Is there any way I can this automatically like reprocess the file if any updates are made on that particular file.

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

-werners- — Thu, 30 Mar 2023 14:54:01 GMT

that sounds more like the change data feed functionality of delta lake.

https://learn.microsoft.com/en-us/azure/databricks/delta/delta-change-data-feed

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

Anonymous — Sat, 01 Apr 2023 02:08:19 GMT

Hi @Sanjay Jain

Hope everything is going great.

Just wanted to check in if you were able to resolve your issue. If yes, would you be happy to mark an answer as best so that other members can find the solution more quickly? If not, please tell us so we can help you.

Cheers!

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

sanjay — Mon, 03 Apr 2023 07:04:14 GMT

Hi Vidula,

Above solutions are not working. Please suggest any other solution.

Regards,

Sanjay