Spark操作OrientDB偶发当前数据库实例未激活错误是什么原因？

• OrientDB会话是线程绑定的，内部通过ThreadLocal将数据库实例和获取它的线程做绑定，一旦跨线程使用其他线程创建的会话实例就会触发该报错。Spark运行在local[*]模式下会启动和本机CPU核心数相等的工作线程，任务调度是随机分配线程执行的，因此冲突发生的概率随机，所以问题表现为偶发不可预测。
• 连接池初始化逻辑错误：你在Driver端初始化了一次连接池后，由于OrientDB连接池无法序列化，无法随Spark任务分发到Executor端，Executor执行时会再次初始化连接池，就是你观察到的对象池被创建两次的现象，不同连接池返回的会话属于不同上下文，会提升会话绑定冲突的概率。
• 会话资源未正确释放：如果会话使用完成后没有主动调用close()方法归还到连接池，会导致会话实例被线程长期持有，后续其他线程复用该会话实例时就会触发线程绑定校验失败。

• 多线程并发场景下，会话使用完不关闭，复用其他线程创建的会话实例，即可偶发该报错。
• 直接跨线程传递会话实例执行数据库操作，可100%触发该报错。
• 同一JVM内初始化多个OrientDB连接池，并发操作下会偶发该报错。

1. 修正连接池初始化时机 #

Executor端连接池按JVM实例只初始化一次，不要在foreach算子中初始化。可采用懒加载单例模式实现DBManager，确保每个Executor的JVM内只会创建一次OrientDB实例和ODatabasePool实例，避免重复创建连接池。

2. 严格遵循会话使用规范 #

每次数据库操作必须遵循「获取会话 -> 执行操作 -> 立即关闭会话」的流程，会话不能跨线程传递，也不能全局缓存复用，操作执行完成后必须立刻关闭会话归还到连接池，示例代码如下：

var session: ODatabaseSession = null
try {
  session = pool.acquire()
  session.command(statement, params)
} catch {
  case t: Throwable => // 自定义异常处理逻辑
} finally {
  if (session != null && session.isActive()) {
    session.close()
  }
}

3. 调整Spark写入逻辑 #

将rdd.foreach替换为rdd.foreachPartition，每个分区只初始化一次连接资源，避免每条数据都重复获取连接，同时减少连接池的压力，调整后代码示例如下：

rdd.repartition(20).foreachPartition(partition => {
  // 每个分区获取一次连接池单例，不会重复初始化
  val pool = DBManager.getPool()
  partition.foreach(row => {
    var session: ODatabaseSession = null
    try {
      session = pool.acquire()
      // 拼接执行数据库操作逻辑
      session.command(statement, params)
    } catch {
      case t: Throwable => // 自定义异常处理逻辑
    } finally {
      if (session != null && session.isActive()) {
        session.close()
      }
    }
  })
})

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