Stage level resource management to handle offheap/onheap memory conflict

[FelixYBW]

Description

As Gluten currently can't fully cover spark functions, we will have to fallback some operators to Vanilla Spark, which leads an conflict of offheap and onheap memory conflict. Currently the suggested solution is that to those queries have both fallback and native operators, we need to config a large offheap memory and a large onheap memory, then the executor number should be decreased due to memory size constraint.

The other solution is to spill offheap memory to onheap memory which we have implemented but performance isn't good, it's not used.

The 3rd solution is to use Spark3.0's stage level resource management, when we detect a stage has fallback, we can fallback the full stage, then restart the executors with high onheap memory.

@zhouyuan @PHILO-HE

[PHILO-HE]

For the 3rd solution, it depends on the stability of our whole stage fallback policy. I will investigate the feasibility. Thanks!

[FelixYBW]

@PHILO-HE have you created a draft PR for this? any insight why the stage level resource manager can't be used by sparkSQL directly?

[SinghAsDev]

Also curious if this can be done in Apache Spark itself, there is some relevant discussion in this SPIP.

[FelixYBW]

@zhli1142015 Here is the related talk. Could you @ the guy so we can sync all here?

[FelixYBW]

Maybe it's known feature. We encountered a case in Spark3.5. Once you enable offheap in Spark, Spark will try to use offheap to cache the internalrows, if offheap memory isn't enough, it won't check the onheap memory but just spill the data. So if you config a large on heap memory but small offheap memory, spill will be triggered more frequently.

While you either can't config a large offheap memory and a small onheap memory because not all operators are able to cache the internal rows into offheap. Onheap demand is still very large for stages fallback to vanilla spark.

[PHILO-HE]

Maybe it's known feature. We encountered a case in Spark3.5. Once you enable offheap in Spark, Spark will try to use offheap to cache the internalrows, if offheap memory isn't enough, it won't check the onheap memory but just spill the data. So if you config a large on heap memory but small offheap memory, spill will be triggered more frequently.

While you either can't config a large offheap memory and a small onheap memory because not all operators are able to cache the internal rows into offheap. Onheap demand is still very large for stages fallback to vanilla spark.

@FelixYBW, we didn't realize this issue before. In Gluten's current resource profile rule, when whole stage fallback occurs, the offheap memory is reduced to 10% of the original value. Then, if I understand correctly, this setting can cause perf regression.

[zhztheplayer]

@PHILO-HE

when whole stage fallback occurs, the off-heap memory is reduced to 10% of the original value.

@FelixYBW

Onheap demand is still very large for stages fallback to vanilla spark.

Not sure if I understand correctly, but are you describing the same phenomenon? Where off-heap usage drops and on-heap usage rises, when the stage falls back?

IIRC Spark sort-based shuffle is a heavy operator that remains on on-heap when off-heap is on. I would be glad to do the some path-findings to see if we can somehow fix this. @FelixYBW If you want to also help locate where the remaining on-heap consumption came from? E.g., Could try setting spark.shuffle.sort.bypassMergeThreshold = 2147483647 to disable vanilla sort-based shuffle then see if the on-heap consumption of vanilla Spark can be reduced. Thanks.

[FelixYBW]

@FelixYBW, we didn't realize this issue before. In Gluten's current resource profile rule, when whole stage fallback occurs, the offheap memory is reduced to 10% of the original value. Then, if I understand correctly, this setting can cause perf regression.

You are right. with whole stage fallback, we may just config onheap memory and disable the offheap memory. 10% offheap memory definitely will lead to more spill.

We need a list of operators which still need large onheap memory when we enable offheap memory.

[FelixYBW]

IIRC Spark sort-based shuffle is a heavy operator that remains on on-heap when off-heap is on. I would be glad to do the some path-findings to see if we can somehow fix this. @FelixYBW If you want to also help locate where the remaining on-heap consumption came from? E.g., Could try setting spark.shuffle.sort.bypassMergeThreshold = 2147483647 to disable vanilla sort-based shuffle then see if the on-heap consumption of vanilla Spark can be reduced. Thanks.

Not the shuffle. From the workload it's the sort operator. We enabled offheap but config a small offheap memory and large onheap memory, the spill is triggered even onheap memory is far enough for the sort.

[FelixYBW]

Since Spark has the unsaferow which can buffer data to offheap. I'd expect that all internalrow cacheing can be save to offheap. But the other struct has to use onheap. Like hashagg or hashjoin, the payload columns are able to save to offheap, but the hash table itself must be saved onheap. @JkSelf do you happen to know when you implement BHJ?

[JkSelf]

@FelixYBW Yes, for BHJ, we need to store the raw data for building the hash table in on-heap memory, and then have the driver distribute it to each executor. Even if we follow Spark's implementation approach of building the hash table only once on the driver side, we still need to store the hash table in on-heap memory before distributing it.