Quick question on memory estimation (on ray). Suppose a simple pipeline that just does a

read_parquet

+ simple

with_column

and

write_parquet

. What would be a good estimation of the memory requirements of a single partition. My full dataset is about

3029 GB

and I have about 24 CPU's I am noticing that I am needing at least 5000 partitions before I hit OOM. With 24 concurrent partitions that's about 15GB of RAM (

3029 / 5000 * 24

). But I have about 180GB of ram on the machine, so I'd expect to handle much bigger partitions. Is there a rule of thumb to estimate, how much

235 GB

of parquet data is going to actually be in memory. Also, for such a pipeline on ray would this data be loaded into the object store, or be in heap memory?

Great questions Parameters to consider: • Ray automatically reserves 30% of your nodes’ memory by default for the object store ◦ Your node thus has about 125GB available ◦ This is about 5GB per core •

3029GB

of parquet is actually much more data in memory, after applying decompression ◦ Assuming a 3x compression ratio, and 5000 partitions, that’s about 1.8GB per partition. ◦ This can be much higher depending on the data modality, and cause much larger inflations at runtime • When determining the amount of memory usage per-partition, you have to consider: ◦ Input partition size (say, roughly 1.8GB per partition) ◦ Intermediate working memory size (this depends on the operation you are running — e.g. url downloads will increase memory usage by a lot) ◦ Output partition size Answers to your questions: 1.

What would be a good estimation of the memory requirements of a single partition.

a. As mentioned we have to take into consideration the input partition size, working memory size and output partition size b. This would be roughly 2-3x the input size (after decompression) depending on what you are running. c. For a simple read -> write, the output partition size is very small (it’s just a dataframe of filepaths) 2. `how much 235 GB of parquet data is going to actually be in memory`: a. This depends on the compression encoding and data (e.g. long strings vs ints). b. Daft does some best-effort guessing here based on the datatypes. c. In the absence of any information, we apply a default of 3x compression ratio (see: _*`parquet_inflation_factor`*_ on our execution config) 3.

Also, for such a pipeline on ray would this data be loaded into the object store, or be in heap memory?

a. For a read -> with_column -> write (with no intermediate

.into_partitions()

), it will be fully pipelined and so wouldn’t touch the object store. b. However, if there is an intermediate

.into_partitions()

call, then the partitions after splitting will be put into the object store before subsequent operations are run on them. i. This is likely your case at the moment since you are splitting your data into 5000 partitions === We are currently working on making this experience much better. Will have more news to share in a few weeks, but certain strategies are available to use here including: 1. Slow ramp-up: schedule one partition per node and accumulate some statistics about how much memory was actually used at runtime vs the input file sizes — we can then use this for subsequent scheduling hints. 2. Better memory usage heuristics: by examining the plan, we can see for example that

.url.download()

is likely to increase memory usage significantly, and we can request for more memory accordingly. 3. Utilizing our new streaming execution engine: this can potentially reduce the memory usage for a given task significantly, and also obviate the need to perform repartitioning, especially for a very streaming-friendly pipeline (with no shuffles) like the ones you are describing All in all, we’re aware of the current pains wrt partitioning and memory. We have good solutions in the works, hope to get your feedback on them soon!

super cool @jay Thanks a lot for the detailed write-up!

Yes — explode does fuse with other operations and should not touch the Ray object store

a. In the absence of any information, we apply a default of 3x compression ratio (see: _*`parquet_inflation_factor`*_ on our execution config)

Hi @jay when exactly does this _*`parquet_inflation_factor`*_ value is used? I tried to set to to higher but the workers still seem to be overscheduled. I set each task size to minimum about 500MB, with the inflation factor of 10 because I expect it to inflate up to 5GB. I see this message here:

Error: No available node types can fulfill resource request {'CPU': 1.0, 'memory': 536890942.0}

Looks like it’s still only asking for ~500MB (instead of 5GB after inflation)

@Henry T by task size you mean min_read_block_size/max_read_block_size?

Our memory estimations need to be improved… I believe the parquet_inflation_factor only kicks in if there are no statistics available. For Parquet, we do retrieve stats and so it doesn’t kick in. Are you trying to hardcode a mechanism for scan sizing? Perhaps we can expose a way, something like user_provided_inflation(…) which will override all other estimations?

I see, then what i observe makes sense. I’m using scan_tasks_min_size_bytes to set the partition size. Do you know if this value is for uncompressed or compressed size?

This value is for the compressed size bytes of the rowgroups. Also note that you may need to set

parquet_split_row_groups_max_files

to be a higher value in order for this splitting to kick in when you have >10 files (which is the current default). --- I’m very sorry about the experience… I know these issues around partition sizing, partition splitting and memory estimations is far from ideal at the moment. I’m compiling a list of pain-points and have a few solutions which I think will help here. Will share a design doc when ready in about ~1 week. The current problems I have identified: • Poor estimations on how much actual memory each task requires • Difficult to resize partitions (especially when you already know that you have really large skews in file sizes in storage) • (Very) large shuffles tend to be expensive and prone to OOMs — we have some prototypes ready here that let us do this already I think we’ll have a much better system in the new year, appreciate your patience! Some potential solutions: • Running our workloads with a “slow ramp-up” time to get a more accurate memory estimation of tasks at runtime, instead of relying purely on planning-time heuristics like we do today ◦ Capturing metrics from historical runs to better inform future runs wrt metrics such as actual data compression ratios, cardinalities, size_bytes etc • Utilizing our new local streaming execution engine to perform workloads in a streaming fashion (very useful especially if your workload has no shuffles)