在复杂的分布式计算环境中，例如基于yarn的spark集群，用户通常通过pyspark客户端连接并提交任务。此时，客户端上安装的pyspark版本可能与集群实际运行的spark core版本不一致。传统的版本检查方法，如pyspark.__version__、ss.version（等同于spark.version）或sc.version，通常只返回pyspark客户端的版本信息，而非集群上spark core的真实版本。即使尝试在用户机器上执行./bin/spark-submit --version，也可能仅显示本地安装的spark提交工具的版本，无法准确反映远程集群的spark core版本。为了解决这一痛点，我们需要一种能够直接查询spark集群运行时版本的方法。
1. 利用Spark SQL version() 函数（Spark 3.0+）
从Spark 3.0版本开始，Spark SQL引入了一个内置的version()函数，可以直接查询当前Spark集群的运行时版本。这个方法是获取Spark Core版本最可靠且通用的方式之一，因为它是在Spark集群上实际执行的SQL查询，因此返回的是集群本身的Spark版本信息。
Java/Scala 示例：
如果你正在使用Java或Scala编写Spark应用程序，可以通过SparkSession执行SQL查询来获取版本：
import org.apache.spark.sql.SparkSession;
public class SparkVersionChecker {
public static void main(String[] args) {
SparkSession spark = SparkSession.builder()
.appName("Spark Core Version Check")
.config("spark.master", "local[*]") // 根据实际环境配置master，例如yarn
.getOrCreate();
// 执行SQL查询获取Spark版本
spark.sql("select version()").show();
spark.stop();
}
}执行上述代码，将得到如下输出（版本号会根据实际集群而异）：
+--------------------+ | version()| +--------------------+ |3.3.2 5103e00c4ce...| +--------------------+
PySpark 示例：
在PySpark中，同样可以通过SparkSession执行SQL查询：
from pyspark.sql import SparkSession
# 假设ss和sc已经通过pyspark.sql.SparkSession.builder连接到集群
# 例如：
# conf = SparkConf().setAppName("SparkVersionChecker").setMaster("yarn")
# ss = SparkSession.builder.config(conf=conf).getOrCreate()
# sc = ss.sparkContext
# 如果你已经有了SparkSession实例，可以直接使用
ss = SparkSession.builder.appName("Spark Core Version Check").getOrCreate()
# 执行SQL查询获取Spark版本
ss.sql("select version()").show(truncate=False)执行上述PySpark代码，同样会输出集群的Spark Core版本：
+----------------------------------------------+ |version() | +----------------------------------------------+ |3.3.2 5103e00c4ce... | +----------------------------------------------+
请注意，truncate=False参数是为了确保完整显示版本字符串，避免被截断。
2. 利用 pyspark.sql.functions.version() 函数（PySpark 3.5+）
对于PySpark 3.5及更高版本，Spark提供了一个更便捷的Python API函数pyspark.sql.functions.version()，它封装了内部的SQL查询逻辑，使得在Python中获取Spark Core版本更加直接和符合Pythonic风格。
PySpark 3.5+ 示例：
from pyspark.sql import SparkSession
from pyspark.sql.functions import version
# 假设ss已经连接到集群
ss = SparkSession.builder.appName("Spark Core Version Check").getOrCreate()
# 创建一个DataFrame，然后使用version()函数
df = ss.range(1) # 创建一个单行DataFrame作为载体
df.select(version()).show(truncate=False)
ss.stop()此方法同样会返回集群的Spark Core版本：
+----------------------------------------------+ |version() | +----------------------------------------------+ |3.5.0 cafbea5b13623276517a9d716f75745eff91f616| +----------------------------------------------+
注意事项与总结
- 版本兼容性： version() SQL函数从Spark 3.0开始可用。pyspark.sql.functions.version()函数则需要PySpark 3.5及以上版本。在选择方法时，请根据你集群和PySpark客户端的实际版本进行选择。
- 执行环境： 上述两种方法的核心优势在于，它们执行的逻辑发生在Spark集群的驱动程序或执行器上，因此能够准确反映集群上部署的Spark Core版本。这与仅仅检查客户端PySpark库版本或本地spark-submit版本有着本质区别。
- 分布式环境： 在YARN、Kubernetes或其他分布式资源管理器上运行Spark时，这种方法尤其有用，因为它能够让你确认远程集群的实际运行版本，这对于调试兼容性问题或确保应用程序在正确版本的Spark上运行至关重要。
- 输出格式： version()函数返回的字符串通常包含版本号和构建哈希值，例如3.3.2 5103e00c4ce...，这提供了非常详细的构建信息。
通过上述方法，你可以可靠地获取分布式Spark集群上实际运行的Spark Core版本，从而更好地管理和维护你的Spark应用程序。
