[VL] GPU and CPU mixed cluster schedule

[jinchengchenghh]

Backend

VL (Velox)

Bug description

We suppose to schedule some IO bound tasks such as the stage contains table scan to CPU node, and some computation intensive tasks to GPU.
Now Spark has this ability to do stage resource scheduler by resource profile as this document https://spark.apache.org/docs/latest/configuration.html#custom-resource-scheduling-and-configuration-overview describes, in Gluten, there has been offheap/onheap memory allocation adjusted by ResourceProfile

This script describes how to set up GPU host environment, the script has executed on the IBM internal AMI linux image, so if you use IBM pipeline pipeline-create-dev-vm and select GPU node such as g4dn.xlarge, the environment is ready, no need to execute the script.
https://raw.githubusercontent.com/jinchengchenghh/gluten/cudf_script/dev/start_cudf_amazon.sh
Note: The environment has been upgraded to cuda 13.1 because cudf build issue, but the script install cuda 12.8, it is outdated.

This document describes how to set up yarn on GPU node.
https://docs.nvidia.com/spark-rapids/user-guide/23.10/getting-started/yarn-gpu.html
https://hadoop.apache.org/docs/stable/hadoop-yarn/hadoop-yarn-site/UsingGpus.html
https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-spark-rapids.html

GPU document describes how to build with GPU
https://github.com/apache/incubator-gluten/blob/main/docs/get-started/VeloxGPU.mdutdated.

Existing offheap/onheap memory ResourceProfile allocation, we should use the similar way to set the profile to require 1 GPU, now the Spark cannot set the core number by resource profile, this feature is under developing.
#8209

We could use TPCDS q95 to test.

The query runs successfully on yarn, but if we set up the environment according to https://docs.nvidia.com/spark-rapids/user-guide/23.10/getting-started/yarn-gpu.html, the query will hang, I also tried stand alone mode before, it also hangs.

Gluten version

No response

Spark version

None

Spark configurations

No response

System information

No response

Relevant logs

[marin-ma]

For data exchange across GPU and CPU clusters, we need to support shuffling data from CPU to GPU and vice versa:

• (mapper stage on CPU cluster) CPU shuffle writer -> (reducer stage on GPU cluster) GPU shuffle reader
• (mapper stage on GPU cluster) GPU shuffle writer -> (reducer stage on CPU cluster) CPU shuffle reader

Currently, we only support cpu shuffle write + cpu shuffle reader, and gpu shuffle writer + gpu shuffle reader, although the output data are in the same serialisation format for cpu and gpu hash shuffle writer. And also there isn't an entry point to tell the shuffle reader whether to output velox rowvector or cudf vector, because the creation of ColumnarShuffleExchangeExec is in the mapper stage, but the decision of scheduling to which cluster is made in the reducer stage.

One feasible approach is to separate the ColumnarShuffleExchangeExec into two operators ColumnarShuffleWriteExec and ColumnarShuffleReadExec, and move the deserialisation from ColumnarShuffleReader.read into ColumnarShuffleReadExec.doExecuteColumnar, then we can set which deserialiser to use in ColumnarShuffleReadExec.

[jinchengchenghh]

Yes, now we don't have the rule to decide if the reducer is GPU because we only have stage level rule, so I assume the gpu exchange is always GPU shuffle writer and GPU shuffle reader. But I have asked @zhztheplayer , injectQueryStagePrepRule can see the total plan, in that case, we can verify if the plan after shuffle reader is supported by cudf backend, if suppported, I will set the cudf tag of the plan to true, and schedule the plan to GPU node.

[marin-ma]

@jinchengchenghh Thanks for your explanation. I'm currently working on this in #11543, and I found that the gpu hash shuffle writer outputs different column data for timestamp and bool type. To support mixed scheduling, the shuffle reader and writer can be scheduled to different clusters. Say if the mappers stage is on gpu cluster but the reducer stage is on cpu cluster, the cpu reader cannot correctly read the data output from gpu writer.

To support this we may also need to have both cpu and gpu shuffle writer output the same format in the shuffle write output. This will introduce some data conversion overhead in gpu shuffle reader for boolean and timestamp type when writer and reader are both gpu.

[jinchengchenghh]

Since we focus on demo TPCDS Q95 which only involves integer data type for exchange, so we reach an agreement that we will defer this discussion, this mismatch may occur in GPU fallback and exchange contains bool and timestamp.

1. Unified writer and separate reader, we can decide if the plan can execute on GPU before this stage executed, collecting some statistic from the last stage, and do some decision
2. GPU writer-> GPU reader, CPU writer -> CPU reader, decide before the plan executes in injectQueryStagePrepRule, eliminate the row by row buffer set of timestamp and bool in shuffle reader.

May decide which is better after TPCH benchmark or some tests specified for bool and timestamp.

For TPCH, we should resolve all GPU fallback after Exchange.