zjuwangg commented on issue #7750: URL: https://github.com/apache/incubator-gluten/issues/7750#issuecomment-2505682053
## Move ColumnarBuildSideRelation's memory occupation to Spark off-heap We are very glad to see the discussion here. In our production environment, we have also been troubled by the out-of-memory (OOM) problem caused by the broadcast build relation using the heap memory for a long time. We adopted a similar approach as proposed by @zhztheplayer and made more optimizations (such as dividing large batches into small batches) in our production scenario. We would like to share our approach and contribute it back in the following weeks. ### Current gluten implement  * Currently, when the ColumnarBuildSideRelation is broadcasted on the driver, the ColumnarBatch is deserialized and stored in the on-heap memory in the data structure `batches: Array[Array[Byte]]`. * On the executor, when the ColumnarBuildSideRelation is constructed, the batches: `Array[Array[Byte]]` still remains in on-heap memory. In some extreme situations, this will consume a large amount of heap memory. Moreover, the batch in `batches: Array[Array[Byte]] ` needs to be copied to off-heap memory through JNI, which will also be a waste of CPU resources and memory. ### Proposed design It went through two rounds of iterative development in our inner environment. #### Round1: using unsafe offheap to store broadcast batches on executor  * Introduce `BytesArrayInOffheap` to store broadcasted data in offheap memory. * On executor side, the broadcast data is first placed on off-heap, and during deserialize process one batch is copied/decoded each time. In this way, the memory occupation of OnHeap has become 1/N of the original (where N is the number of batches). However, it is obvious that there is room for improvement to avoid the extra copying between on-heap and off-heap memory. Additionally, another problem emerges where a certain batch in `batches: Array[Array[Byte]]` can be extremely large, which usually leads to out-of-memory (OOM) in off-heap memory. Consequently, we carried out our second round of optimization. #### Round2: avoid extra copy between heap/offheap and serialize more small batched to construct `batches: Array[Array[Byte]]`  * In `BroadcastUtils#serializeStream`, support split `batches: Iterator[ColumnarBatch]` into more small batches * On executor side, the broadcast data is directly read from offheap. We pass one batch in `BytesArrayInOffheap` memory address and size to underlying deserializer. ### Implement Steps We will introudce a config in GlutenConfig which controls whether use offheap to store the broadcast data. And when all related code get merged and after all things be done, we can remove the config and make this as default behavior. Briefly Steps: * Introduce `BytesArrayInOffheap`、 `UnsafeColumnarBuildSideRelation` and `spark.gluten.sql.BroadcastBuildRelationUseOffheap.enabled` in GlutenConfig to Implement what we have done in Step1 * Add more interface `BroadcastUtils#serializeStream` and refactor the deserialize/serialize the process * support split single large batch into more small batche * support direct read from offheap when doing deserialize in native code * Make the broadcast using offheap as the default behavior. cc @WangGuangxin @weiting-chen -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
