Hi Team CC: @Everett Kleven @Cory Grinstead @Cory Grinstead Some findings from my end, after playing around with Daft for our needs. As of today most of my pipelines start as: 1. ray.data.read_databricks_tables: using SQL syntax 2. convert to daft 3. Daft at this point is usually 1 video per row. 4. I later explode videos to images (300 frames per video) 5. I then download and decode the videos (using ray-native functions) 6. I then do preprocessing (using my own custom preprocessing functions) 7. I then do inference on different kinds of models some huge some tiny. Usually 1 model at a time. 8. I then write to parquet (append): where every image is a row Findings: 1. I always need to do into_batches immediately after explode in order to ensure the RAM doesn't get fill. a. I tried before the explode with lower values. I tried removing the into_batches and using dynamic batches, but so only into_batches immediately after explode works. b. This is usually because without it the temporary

preprocessed_image

(gets deleted after inference) just creates too many numpy tensors that the model cannot consume fast enough. 2. I also have switch from using daft.func for preprocessing to using daft.cls in order to always make the max_concurrency of the preprocessing be bounded to 2 * num_gpus. a. All my preprocessing are now using daft.cls with max_concurrency=num_gpus *2 b. It's my hacky way to control the queue size between preprocessing -> inference from exploding. c. this is somewhat equivalent to prefetct_factor=2 in Pytorch Dataloaders. 3. For very big models that have slow images per second I am also having to throttle the max_connection of daft download function a. why? to again ensure queue size between download + decode -> preprocess_udf does not explode. That's 3 knobs I'm having to control differently per each model or pipeline in order from not having one step in the pipeline producing a queue too large for the next step. Which makes me wonder is there is a better solution for this or if we could have more control of the queue size or not slow down one steps if it's to fast for the next one? Some possible solution: The max_in_flight or queue_size can just be a parameter for all native or user-defined daft.func and daft.cls.

df = df.with_column("image_bytes", df["image_url"].url.download()).max_in_flight(64)
df = df.with_column("image", df["image_bytes"].decode_image()).max_in_flight(32)
df = df.with_column("preprocessed_image", preprocessing_udf(col("image"))).max_in_flight(16)
df = df.select("*", inference_udf(...)).max_in_flight(8)

The queue numbers are number of batches not number of rows.
[download]──[64]──[decode]──[32]──[preprocess]──[16]──[infer]──[8]──[write]
              ↑               ↑                  ↑              ↑
        image_bytes        image          preprocessed_image  predictions
        rows queued     rows queued         rows queued      rows queued

Let me know what you guys think, hopefully this helps.

@Abner Ayala Have you tried manually setting the

default_morsel_size

. By default we use a very large morsel (queue) size. Adjusting this would reduce the global "queue" size used for every operator.

daft.set_execution_config(default_morsel_size=size)

thanks, I just tried it, but it doesn't eliminate the other knobs, it mainly creates a new one morsel_size so that I can make the into_batches a bit bigger. So the main advantage is it allows me to make into_batches after the explode bigger, but I still need to set all the other knobs or else it will OOM.

I believe daft should perform backpressure already, not sure if we have the mechanisms to control the size of the queue between operators. Someone else would be a better candidate to chime in, perhaps @Colin Ho

Thanks @jay I'll await the teams suggestions. I believe there are some mechanism to control the some of the functions speed like upload and download max_connections and max_concurrency in

daft.cls

. The main thing is not like all operations have a mechanisms to enforce them to slow down. Most of my issues with daft that have made it a bit tedious is there is no slow down option. Like if ops 1 is way faster than ops 2 and ops1 output has a big object (tensor) it just explodes and start filling the OOM or the object storage depending of pipeline setup. My suggestion about controlling the queue is more like ops1 already fills the output buffer it should just block or slow down until it needs to fill the buffer again. This pattern keeps happening across my pipelines. Like yesterday I had a pipeline that just read a large parquet did some very light calculation and then used daft.upload to upload a bin blob per row and when I ran on few rows it work perfectly on the first try but when running at scale I had to reduce the max_connections to avoid s3 PUT limits and then I had to add a into_partitions(small number) after the read_parquet and I had to change some of the execute config values. So 3 different knobs I needed to change just to tell daft please slow down all the other ops are running blazing fast but we cannot just do a billion s3 PUT at once. Somehow it needs to be smarter about when to slow down so things are more plug and play and we don't need magic number or configs to get things to just work.