[GLUTEN-11524][VL][WIP] Support adjusting stage execution mode

[marin-ma]

wip

Related issue: #11524

[github-actions]

Run Gluten Clickhouse CI on x86

[marin-ma]

Steps to test this feature in a local standalone spark cluster with mocked GPU resources. No actual GPU resources are required.

1. Create resource scripts for CPU workers and GPU workers

• CPU workers

  cpu.conf

  spark.worker.resource.cpu.amount	1
  spark.worker.resource.cpu.discoveryScript /path/to/cpu.sh
  

  cpu.sh

  #!/usr/bin/env bash
  
  echo {\"name\": \"cpu\", \"addresses\":[\"0\"]}
  

• GPU workers

  gpu.conf

  spark.worker.resource.gpu.amount	1
  spark.worker.resource.gpu.discoveryScript /path/to/gpu.sh
  

  gpu.sh

  #!/usr/bin/env bash
  
  echo {\"name\": \"gpu\", \"addresses\":[\"1\"]}
  

2. Start spark master, 2 CPU workers and 1 GPU workers. Each CPU workers has one "cpu" resource, and each GPU workers has one "gpu" resource.

sbin/start-master.sh

export SPARK_WORKER_DIR=/tmp/spark-worker-1
export SPARK_PID_DIR=/tmp/spark-pid-1
./sbin/start-worker.sh spark://localhost:7077 \
  --webui-port 8081 --properties-file /path/to/gpu.conf
 
 
export SPARK_WORKER_DIR=/tmp/spark-worker-2
export SPARK_PID_DIR=/tmp/spark-pid-2
./sbin/start-worker.sh spark://localhost:7077 \
  --webui-port 8082 --properties-file /path/to/cpu.conf
 
export SPARK_WORKER_DIR=/tmp/spark-worker-3
export SPARK_PID_DIR=/tmp/spark-pid-3
./sbin/start-worker.sh spark://localhost:7077 \
  --webui-port 8083 --properties-file /path/to/cpu.conf

3. Run spark application with below configurations added. This will start 1 executor on each worker node. Default execution will be scheduled onto CPU worker nodes, and the selected gpu stages (configured by spark.gluten.auto.adjustStageExecutionMode=true, currently only the join stages) will be scheduled onto GPU worker nodes. If set spark.gluten.auto.adjustStageExecutionMode=false, all stages that can use cudf will be scheduled onto GPU worker nodes.

spark.driver.extraJavaOptions "-Dspark.testing=true -Dio.netty.tryReflectionSetAccessible=true"
spark.dynamicAllocation.enabled true
spark.executor.resource.cpu.amount 1
spark.executor.resource.cpu.discoveryScript=/path/to/cpu.sh
spark.gluten.sql.columnar.backend.velox.cudf.enableValidation=false
spark.gluten.sql.columnar.cudf=true
spark.gluten.auto.adjustStageExecutionMode=true

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[marin-ma]

Verified with tpcds q95 on a gpu node with below configurations:

cat tpcds_parquet.scala | ${SPARK_HOME}/bin/spark-shell \
  --master spark://5da02429f127:7077 --deploy-mode client \
  --executor-cores 4 \
  --conf spark.plugins=org.apache.gluten.GlutenPlugin \
  --conf spark.driver.extraClassPath=${GLUTEN_JAR} \
  --conf spark.executor.extraClassPath=${GLUTEN_JAR} \
  --conf spark.memory.offHeap.enabled=true \
  --conf spark.memory.offHeap.size=2g \
  --conf spark.gluten.sql.columnar.forceShuffledHashJoin=true \
  --conf spark.shuffle.manager=org.apache.spark.shuffle.sort.ColumnarShuffleManager \
  --conf spark.driver.extraJavaOptions="-Dio.netty.tryReflectionSetAccessible=true -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=*:5005" \
  --conf spark.executor.extraJavaOptions="-Dio.netty.tryReflectionSetAccessible=true" \
  --conf spark.dynamicAllocation.enabled=true \
  --conf spark.executor.resource.cpu.amount=1 \
  --conf spark.executor.resource.cpu.discoveryScript=/opt/spark/conf/cpu.sh \
  --conf spark.gluten.sql.columnar.backend.velox.cudf.enableValidation=true \
  --conf spark.gluten.sql.columnar.cudf=true \
  --conf spark.gluten.sql.debug=true \
  --conf spark.gluten.sql.debug.cudf=true \
  --conf spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleOutput=true \
  --conf spark.gluten.sql.columnar.backend.velox.resizeBatches.shuffleInput=false \
  --conf spark.gluten.auto.adjustStageExecutionMode=true \
  --conf spark.log.level=INFO

Started 2 cpu workers and 1 gpu worker using the script above. During runtime there's one executor started on each worker. Only the 40 tasks of stage 40 were scheduled to the executor on the gpu worker, all other tasks were scheduled to the executor on the 2 cpu workers. Next will try it on a real mixed cpu/gpu cluster.

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86