Spark 如何处理涉及 JDBC 数据源的故障场景？

Question

我正在编写一个与 Spark 的 JDBC 数据源实现有相似之处的数据源，我想问一下 Spark 如何处理某些故障场景。据我了解，如果执行器在运行任务时死亡，Spark 将恢复执行器并尝试重新运行该任务。然而，这在数据完整性和 Spark 的 JDBC 数据源 API（例如df.write.format("jdbc").option(...).save()）的上下文中如何发挥作用？

在JdbcUtils.scala的savePartition函数中，我们看到Spark调用了用户提供的数据库url/凭据生成的Java连接对象的提交和回滚函数（见下文）。但是，如果执行器在 commit() 完成后或调用 rollback() 之前立即死亡，Spark 是否会尝试重新运行任务并再次写入相同的数据分区，实质上是在数据库中创建重复的已提交行？如果 executor 在调用 commit() 或 rollback() 的过程中死亡会发生什么？

try {
    ...
    if (supportsTransactions) {
        conn.commit()
    }
    committed = true
    Iterator.empty
} catch {
    case e: SQLException =>
        ...
        throw e
} finally {
    if (!committed) {
        // The stage must fail.  We got here through an exception path, so
        // let the exception through unless rollback() or close() want to
        // tell the user about another problem.
        if (supportsTransactions) {
          conn.rollback()
        }
        conn.close()
    } else {
        ...
    }
}

Answer 1

但是，如果执行器在 commit() 完成后或调用 rollback() 之前立即死亡，Spark 是否会尝试重新运行任务并再次写入相同的数据分区，从而在数据库中创建重复的已提交行？

既然 Spark SQL（它是一种基于 RDD API 的高级 API）对 JDBC 或任何其他协议的所有特性知之甚少，您会期待什么？更不用说底层执行运行时，即 Spark Core。

当您编写像 df.write.format(“jdbc”).option(...).save() 这样的结构化查询时，Spark SQL 会使用类似程序集的低级 RDD API 将其转换为分布式计算。由于它试图包含尽可能多的“协议”（包括 JDBC），Spark SQL 的 DataSource API 将大部分错误处理留给了数据源本身。

调度任务（不知道甚至不关心任务做什么）的 Spark 核心只是监视执行，如果任务失败，它将尝试再次执行它（直到默认尝试 3 次失败）。

因此，当您编写自定义数据源时，您会知道如何进行练习，并且必须在代码中处理此类重试。

处理错误的一种方法是使用TaskContext（例如addTaskCompletionListener 或addTaskFailureListener）注册任务侦听器。

Answer 2

出于所描述的原因，我不得不引入一些重复数据删除逻辑。实际上，您最终可能会两次（或更多）提交相同的内容。