如何通过名称获取已注册的Spark Accumulator而无需传递实际引用？

其实Spark本身并没有直接提供getAccumulatorByName这种开箱即用的API，但我们可以通过自定义一个简单的管理工具来实现你想要的效果，下面给你具体的方案：

1. 自定义Accumulator管理单例 #

我们可以写一个单例对象来统一管理所有注册的Accumulator，在创建Accumulator时同时将它存入一个名称到实例的映射表，之后就可以通过名称直接检索：

import org.apache.spark.SparkContext
import org.apache.spark.util.LongAccumulator

object AccumulatorManager {
  // 用可变Map存储名称与Accumulator的映射
  private val accumulatorMap = scala.collection.mutable.Map[String, Any]()

  // 封装LongAccumulator的创建与注册逻辑
  def createAndRegisterLongAcc(sc: SparkContext, name: String): LongAccumulator = {
    val acc = sc.longAccumulator(name)
    accumulatorMap.put(name, acc)
    acc
  }

  // 通过名称获取Accumulator的方法
  def getAccumulatorByName(name: String): Option[Any] = {
    accumulatorMap.get(name)
  }
}

2. 使用示例 #

按照你的预期场景，使用方式如下：

// 创建并注册名为cnt1的LongAccumulator
val cnt1 = AccumulatorManager.createAndRegisterLongAcc(sc, "cnt1")
// 通过名称获取实例并转换类型
val cnt2 = AccumulatorManager.getAccumulatorByName("cnt1").asInstanceOf[LongAccumulator]

cnt1.add(1)
println(cnt2.value) // 输出1，和预期一致

3. 额外的安全建议 #

直接用asInstanceOf做类型转换有抛出异常的风险，更安全的做法是用模式匹配做类型检查：

AccumulatorManager.getAccumulatorByName("cnt1") match {
  case Some(acc: LongAccumulator) => println(s"获取到Accumulator，值为：${acc.value}")
  case Some(_) => println("找到的Accumulator类型不匹配")
  case None => println("未找到对应名称的Accumulator")
}

另外要注意，这个管理类是运行在Driver端的，因为Accumulator本质是Driver端维护的对象，Executor端只能更新它的值，不能直接获取引用，所以确保所有注册和获取操作都在Driver端执行即可。

需要说明的是，Spark原生的SparkContext并没有对外暴露通过名称检索Accumulator的接口，内部虽然会跟踪已注册的Accumulator，但这些信息并没有开放给用户使用，所以自定义管理是最直接的解决方案。

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