Spark版本升级后如何通过反射调用更名的transformDown相关方法？

我基于Spark 3.1.2编写的代码如下：

private def work(plan: LogicalPlan): LogicalPlan = {
  val result = plan.transformDown {
    // 无关业务逻辑
  }
}

将这段代码放到Spark 3.3.0环境运行时，抛出如下错误：

java.lang.NoSuchMethodError: org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDown(Lscala/PartialFunction;)Lorg/apache/spark/sql/catalyst/plans/logical/LogicalPlan;

原因是Spark 3.3.0中已移除transformDown方法，替换为transformDownWithPruning方法。

我希望实现跨版本兼容逻辑，不想硬编码版本判断，而是通过反射**按方法名包含"transformDown"**自动匹配调用对应方法。目前已写出部分反射代码：

private def transformWithReflection(plan: LogicalPlan) = {
    val runtime = scala.reflect.runtime.universe
    val mirror = runtime.runtimeMirror(getClass.getClassLoader)
    val instanceMirror = mirror.reflect(plan)

    // 这里是精确查找transformDown，不符合按名称包含筛选的需求
    val transformMethodAlternatives = runtime
      .typeOf[LogicalPlan]
      .decl(runtime.TermName("transformDown"))
      .asTerm
      .alternatives
    
    ...
    // 待实现调用反射方法的逻辑
}

请问如何实现过滤出名称包含"transformDown"的方法并调用？或者有没有更合理的兼容方案？

方案1：反射筛选名称包含"transformDown"的方法 #

可以遍历LogicalPlan的所有公开方法，筛选出名称包含"transformDown"且参数匹配PartialFunction[LogicalPlan, LogicalPlan]的方法，再通过反射调用。这种方式无需硬编码版本，自动适配方法名变化：

import scala.reflect.runtime.universe
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan

private def transformWithReflection(plan: LogicalPlan, pf: PartialFunction[LogicalPlan, LogicalPlan]): LogicalPlan = {
  val runtime = universe
  val mirror = runtime.runtimeMirror(getClass.getClassLoader)
  val instanceMirror = mirror.reflect(plan)
  val planType = runtime.typeOf[LogicalPlan]

  // 筛选符合条件的方法：名称含transformDown、参数为PartialFunction、是方法类型
  val targetMethod = planType.decls
    .filter(_.isTerm)
    .map(_.asTerm)
    .filter(term => 
      term.name.toString.contains("transformDown") && 
      term.isMethod && 
      term.paramLists.flatten.size == 1 && 
      term.paramLists.flatten.head.typeSignature =:= runtime.typeOf[PartialFunction[LogicalPlan, LogicalPlan]]
    )
    .headOption
    .getOrElse(throw new NoSuchMethodError("未找到匹配的transformDown相关方法"))

  // 调用反射方法并转换返回值类型
  instanceMirror.reflectMethod(targetMethod)(pf).asInstanceOf[LogicalPlan]
}

// 使用示例
private def work(plan: LogicalPlan): LogicalPlan = {
  transformWithReflection(plan, {
    // 原transformDown中的业务处理逻辑
  })
}

方案2：版本判断+精确反射调用 #

如果担心后续Spark版本出现多个含"transformDown"的重载方法，可先通过Spark版本判断，再精确调用对应方法，稳定性更高：

import scala.reflect.runtime.universe
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan

// 获取当前Spark版本
private def getSparkVersion: String = SparkSession.getActiveSession.map(_.version).getOrElse("unknown")

private def transformWithVersionCheck(plan: LogicalPlan, pf: PartialFunction[LogicalPlan, LogicalPlan]): LogicalPlan = {
  val runtime = universe
  val mirror = runtime.runtimeMirror(getClass.getClassLoader)
  val instanceMirror = mirror.reflect(plan)
  
  // 根据版本选择方法名
  val methodName = if (getSparkVersion.startsWith("3.1")) "transformDown" else "transformDownWithPruning"
  
  // 精确查找并调用方法
  val targetMethod = runtime.typeOf[LogicalPlan]
    .decl(runtime.TermName(methodName))
    .asTerm
    .alternatives
    .filter(_.asMethod.paramLists.flatten.size == 1)
    .head
    .asMethod

  instanceMirror.reflectMethod(targetMethod)(pf).asInstanceOf[LogicalPlan]
}

注意事项 #

• 反射调用必须严格匹配参数类型，否则会抛出IllegalArgumentException
• 若后续Spark版本新增同名重载方法，需补充返回值类型等筛选条件
• 优先选择版本判断方案，稳定性更强；名称筛选方案适合快速适配已知版本差异

内容的提问来源于stack exchange，提问作者stackoverflowflowflwofjlw