Hi, daft team. Is there any backpressure mechanism for pipeline to avoid OOM? for example I have a image download UDF which download image very fast, but the second UDF process each image for 20seconds. Then the image I previously downloaded (to memory) will cost a lot of memory usage. Is there any option to allow the first UDF to "slow-down" when second one is already overwhelming?

Yeah you can specify a smaller

batch_size

on image download UDF, and a larger one on the image processing UDF

I see, so let's say if ImageDownloadUDF has batch size 100, Image Processing has batch size 10000, which means the Image Processing will not start to operate before the ImageDownload proceed 10000 records? Will this further increase the burden of ImageDownload memory usage?

Image Processing will not start to operate before the ImageDownload proceed 10000 records?

This is correct. But, the ImageDownload will pause once the ImageProcess UDF is running.

What if I also change ImageProcess to a very small batch? for example same as. ImageDownload = 100?

That should also work

Got it, but we saw that currently it reaches OOM for this kind of setting, using daft 0.6.6, is there any hint for this? (Sorry I cannot share error log / code here)

What if you set it to very small, like batch_size=1

Let me try, I want some concurrency so I will set cpu to 0.1 (I only have 1 machine), concurrency = 100 (so it use 10 CPUs), with a very low batch size

it depends on the volume of data, but generally yes because it is a fully materializing operation. However, it is able to spill to disk if needed

Got it~ Thanks for the support!

FWIW, we are working on a new dynamic batching algorithm for UDF's that will allow it to automatically adjust based off runtime metrics!