问题

进行Gluten+Velox在Spark on YARN集群的TPCH/TPCDS测试时，通过在spark-shell提交任务，将Spark任务以YARN-client模式，Spark客户端节点将Spark任务提交至YARN集群运行。

运行过程中，报错ClassNotFound org.apache.spark.shuffle.sort.ColumnarShuffleManager：

[2023-09-06 10:56:45.063]Container exited with a non-zero exit code 1. Error file: prelaunch.err.
Last 4096 bytes of prelaunch.err :
Last 4096 bytes of stderr :
...
Exception in thread "main" java.lang.reflect.UndeclaredThrowableException
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1761)
at org.apache.spark.deploy.SparkHadoopUtil.runAsSparkUser(SparkHadoopUtil.scala:61)
at org.apache.spark.executor.CoarseGrainedExecutorBackend$.run(CoarseGrainedExecutorBackend.scala:419)
at org.apache.spark.executor.YarnCoarseGrainedExecutorBackend$.main(YarnCoarseGrainedExecutorBackend.scala:81)
at org.apache.spark.executor.YarnCoarseGrainedExecutorBackend.main(YarnCoarseGrainedExecutorBackend.scala)
Caused by: java.lang.ClassNotFoundException: org.apache.spark.shuffle.sort.ColumnarShuffleManager
at java.net.URLClassLoader.findClass(URLClassLoader.java:387)
at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:352)
at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
at java.lang.Class.forName0(Native Method)
at java.lang.Class.forName(Class.java:348)
at org.apache.spark.util.Utils$.classForName(Utils.scala:216)
at org.apache.spark.util.Utils$.instantiateSerializerOrShuffleManager(Utils.scala:2671)
at org.apache.spark.SparkEnv$.create(SparkEnv.scala:315)
at org.apache.spark.SparkEnv$.createExecutorEnv(SparkEnv.scala:207)
at org.apache.spark.executor.CoarseGrainedExecutorBackend$.$anonfun$run$7(CoarseGrainedExecutorBackend.scala:468)
at org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:62)
at org.apache.spark.deploy.SparkHadoopUtil$$anon$1.run(SparkHadoopUtil.scala:61)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:422)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1746)
... 4 more

这实际是gluten编译的jar包依赖在Worker节点没有找到，但在提交任务的配置中，明明看似已经指定了所需jar依赖的路径，因此有必要搞清，尤其是在集群环境下的Spark的依赖加载原理。

Spark依赖加载

Spark 为开发者提供的主要是 3 种配置方式：配置文件（spark-defaults.conf）、命令行参数和 SparkConf 对象。这些方式的特点都是以（K，V）键值对的形式表示的，设计了三种配置方式自然是为了满足不同的应用场景，总的来说，配置文件的方式是应用于全局，而Spark命令行参数和 SparkConf 对象的方式是应用级别的配置。

实际运行时，对于这些多处指定的配置项，Spark会按照 SparkConf对象 -> 命令行-conf参数 -> 配置文件 的顺序，依次读取配置项的参数值，重复配置项以先读到的值为准。但尽管如此，实际使用时，最好明确每项配置的作用并仅在相应位置出现，这便于开发者和阅读者的理解。

官方文档中，在提交任务的命令行使用-conf，通过spark.driver.extraClassPath、spark.executor.extraClassPath指定Gluten jar所在的路径。

export GLUTEN_JAR = /PATH/TO/GLUTEN/backends-velox/target/<gluten-jar>
export SPARK_HOME =/PATH_TO_SPARK_HOME/
cat tpch_parquet.scala | ${SPARK_HOME}/bin/spark-shell \
  --master yarn --deploy-mode client \ 
..
  --conf spark.driver.extraClassPath=${GLUTEN_JAR} \
  --conf spark.executor.extraClassPath=${GLUTEN_JAR} 
..

将GLUTEN_JAR 指定为Gluten路径编译打包的target路径，就出现了文章一开始的错误。

涉及到引入JAR依赖时，配制文件及命令行可用的配置包括：

• CLASSPATH
  • spark.driver.extraClassPath or --driver-class-path 指定driver节点上额外的classpath去寻找依赖
  • spark.executor.extraClassPath 同上，但指定的是worker节点上的classpath

• spark.yarn.
  • spark.yarn.jars
  • spark.yarn.archive              
• --jars
• spark.driver.extraLibraryPath

CLASSPATH的配置方式如果需要在master和worker节点均生效，则两项均需配置。extraLibraryPath的配置指示的是一个用于driver 在JVM启动时就能去加载的依赖位置，因此其没有在代码中通过SparkConf设置的方法（因为JVM已启动）。

在Spark集群环境中主要关注这几种配置方法能否实现集群间依赖的“分发”。在Spark on YARN环境配置时仅靠Spark客户端节点指定HADOOP_CONF_DIR或YARN_CONF_DIR以告知YARN的配置即可，因此Spark有自身的依赖分发。

通过打印INFO级别的日志：

...
23/09/26 12:29:53 INFO Client: Requesting a new application from cluster with 3 NodeManagers
23/09/26 12:29:54 INFO Client: Verifying our application has not requested more than the maximum memory capability of the cluster (8192 MB per container)
23/09/26 12:29:54 INFO Client: Will allocate AM container, with 896 MB memory including 384 MB overhead
23/09/26 12:29:54 INFO Client: Setting up container launch context for our AM
23/09/26 12:29:54 INFO Client: Setting up the launch environment for our AM container
23/09/26 12:29:54 INFO Client: Preparing resources for our AM container
23/09/26 12:29:54 WARN Client: Neither spark.yarn.jars nor spark.yarn.archive is set, falling back to uploading libraries under SPARK_HOME.
23/09/26 12:30:10 INFO Client: Uploading resource file:/tmp/spark-86cfe1cf-b976-4d15-b4ac-1aeb03210f15/__spark_libs__8828641869057990246.zip -> hdfs://mycluster/user/johndoe/.sparkStaging/application_1695694421109_0001/__spark_libs__8828641869057990246.zip
23/09/26 12:30:19 INFO Client: Uploading resource file:/tmp/spark-86cfe1cf-b976-4d15-b4ac-1aeb03210f15/__spark_conf__2680388579298625063.zip -> hdfs://mycluster/user/johndoe/.sparkStaging/application_1695694421109_0001/__spark_conf__.zip
...

结论

Neither spark.yarn.jars nor spark.yarn.archive is set, falling back to uploading libraries under SPARK_HOME. 意味着Spark实现依赖在集群节点间的分发动作是将SPARK_HOME/jars路径下的所有依赖打包上传至HDFS从而使其他节点获取依赖。 

该行日志带出的spark.yarn.jars/archives配置项也负责依赖的集群间共享，而该配置启用后将不再上传SPARK_HOME下的/jars。而且spark.yarn.jars配置项完全可以指定为一个HDFS路径——在启动时，Spark将认为依赖已放在目的路径，这避免了每次启动时较为耗时的上传操作。

CLASSPATH本身并不带有指示分发的含义，要使依赖正确被集群获取，要么使集群中每个节点都拥有CLASSPATH指定的本地路径，要么将所需的jar放在SPARK_HOME/jars或spark.yarn.jars的指定路径，利用启动时的上传分发至集群。

--jars的方式需要指定集群各节点都能访问到的路径，且一般建议用于额外依赖较少的情况。

最后，spark.driver.extraLibraryPath目前主要用在了一些本地库(.so)的加载

目前采用的方式是将Gluten编译的jar包放置于SPARK_HOME/jars下。