kyuubi spark lineage集成Atlas

Question

在使用apache kyuubi spark lineage插件集成Atlas过程中，出现报错，我使用版本是spark 3.5.3+Atlas 2.4.0
报错内容如下：

```shell
26/02/06 09:17:54 WARN AtlasLineageDispatcher: Send lineage to atlas failed.
java.lang.IllegalStateException: Shutdown in progress, cannot add a shutdownHook
    at org.apache.hadoop.util.ShutdownHookManager.addShutdownHook(ShutdownHookManager.java:301) ~[hadoop-client-api-3.4.0-amzn-2.jar:?]
    at org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClient$.registerCleanupShutdownHook(AtlasClient.scala:88) ~[kyuubi-spark-lineage_2.12-1.11.0.jar:1.11.0]
    at org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasClient$.getClient(AtlasClient.scala:74) ~[kyuubi-spark-lineage_2.12-1.11.0.jar:1.11.0]
    at org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasLineageDispatcher.$anonfun$send$2(AtlasLineageDispatcher.scala:37) ~[kyuubi-spark-lineage_2.12-1.11.0.jar:1.11.0]
    at org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasLineageDispatcher.$anonfun$send$2$adapted(AtlasLineageDispatcher.scala:30) ~[kyuubi-spark-lineage_2.12-1.11.0.jar:1.11.0]
    at scala.Option.foreach(Option.scala:407) ~[scala-library-2.12.18.jar:?]
    at org.apache.kyuubi.plugin.lineage.dispatcher.atlas.AtlasLineageDispatcher.send(AtlasLineageDispatcher.scala:30) ~[kyuubi-spark-lineage_2.12-1.11.0.jar:1.11.0]
    at org.apache.kyuubi.plugin.lineage.SparkOperationLineageQueryExecutionListener.$anonfun$onSuccess$1(SparkOperationLineageQueryExecutionListener.scala:35) ~[kyuubi-spark-lineage_2.12-1.11.0.jar:1.11.0]
    at org.apache.kyuubi.plugin.lineage.SparkOperationLineageQueryExecutionListener.$anonfun$onSuccess$1$adapted(SparkOperationLineageQueryExecutionListener.scala:35) ~[kyuubi-spark-lineage_2.12-1.11.0.jar:1.11.0]
    at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62) ~[scala-library-2.12.18.jar:?]
    at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55) ~[scala-library-2.12.18.jar:?]
    at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49) ~[scala-library-2.12.18.jar:?]
    at org.apache.kyuubi.plugin.lineage.SparkOperationLineageQueryExecutionListener.onSuccess(SparkOperationLineageQueryExecutionListener.scala:35) ~[kyuubi-spark-lineage_2.12-1.11.0.jar:1.11.0]
    at org.apache.spark.sql.util.ExecutionListenerBus.doPostEvent(QueryExecutionListener.scala:173) ~[spark-sql_2.12-3.5.3-amzn-0.jar:3.5.3-amzn-0]
    at org.apache.spark.sql.util.ExecutionListenerBus.doPostEvent(QueryExecutionListener.scala:143) ~[spark-sql_2.12-3.5.3-amzn-0.jar:3.5.3-amzn-0]
    at org.apache.spark.util.ListenerBus.postToAll(ListenerBus.scala:117) ~[spark-core_2.12-3.5.3-amzn-0.jar:3.5.3-amzn-0]
    at org.apache.spark.util.ListenerBus.postToAll$(ListenerBus.scala:101) ~[spark-core_2.12-3.5.3-amzn-0.jar:3.5.3-amzn-0]
    at org.apache.spark.sql.util.ExecutionListenerBus.postToAll(QueryExecutionListener.scala:143) ~[spark-sql_2.12-3.5.3-amzn-0.jar:3.5.3-amzn-0]
    at org.apache.spark.sql.util.ExecutionListenerBus.onOtherEvent(QueryExecutionListener.scala:155) ~[spark-sql_2.12-3.5.3-amzn-0.jar:3.5.3-amzn-0]
    at org.apache.spark.scheduler.SparkListenerBus.doPostEvent(SparkListenerBus.scala:100) ~[spark-core_2.12-3.5.3-amzn-0.jar:3.5.3-amzn-0]
    at org.apache.spark.scheduler.SparkListenerBus.doPostEvent$(SparkListenerBus.scala:28) ~[spark-core_2.12-3.5.3-amzn-0.jar:3.5.3-amzn-0]
    at org.apache.spark.scheduler.AsyncEventQueue.doPostEvent(AsyncEventQueue.scala:37) ~[spark-core_2.12-3.5.3-amzn-0.jar:3.5.3-amzn-0]
    at org.apache.spark.scheduler.AsyncEventQueue.doPostEvent(AsyncEventQueue.scala:37) ~[spark-core_2.12-3.5.3-amzn-0.jar:3.5.3-amzn-0]
    at org.apache.spark.util.ListenerBus.postToAll(ListenerBus.scala:117) ~[spark-core_2.12-3.5.3-amzn-0.jar:3.5.3-amzn-0]
    at org.apache.spark.util.ListenerBus.postToAll$(ListenerBus.scala:101) ~[spark-core_2.12-3.5.3-amzn-0.jar:3.5.3-amzn-0]
    at org.apache.spark.scheduler.AsyncEventQueue.super$postToAll(AsyncEventQueue.scala:105) ~[spark-core_2.12-3.5.3-amzn-0.jar:3.5.3-amzn-0]
    at org.apache.spark.scheduler.AsyncEventQueue.$anonfun$dispatch$1(AsyncEventQueue.scala:105) ~[spark-core_2.12-3.5.3-amzn-0.jar:3.5.3-amzn-0]
    at scala.runtime.java8.JFunction0$mcJ$sp.apply(JFunction0$mcJ$sp.java:23) ~[scala-library-2.12.18.jar:?]
    at scala.util.DynamicVariable.withValue(DynamicVariable.scala:62) ~[scala-library-2.12.18.jar:?]
    at org.apache.spark.scheduler.AsyncEventQueue.org$apache$spark$scheduler$AsyncEventQueue$$dispatch(AsyncEventQueue.scala:100) ~[spark-core_2.12-3.5.3-amzn-0.jar:3.5.3-amzn-0]
    at org.apache.spark.scheduler.AsyncEventQueue$$anon$2.$anonfun$run$1(AsyncEventQueue.scala:96) ~[spark-core_2.12-3.5.3-amzn-0.jar:3.5.3-amzn-0]
    at org.apache.spark.util.Utils$.tryOrStopSparkContext(Utils.scala:1358) [spark-core_2.12-3.5.3-amzn-0.jar:3.5.3-amzn-0]
    at org.apache.spark.scheduler.AsyncEventQueue$$anon$2.run(AsyncEventQueue.scala:96) [spark-core_2.12-3.5.3-amzn-0.jar:3.5.3-amzn-0]

```

Answer 1

下午好🌅🌅🌅
本答案参考通义千问

你遇到的问题是在使用 Apache Kyuubi 的 Spark Lineage 插件 集成 Apache Atlas 时，出现了以下错误：

java.lang.IllegalStateException: Shutdown in progress, cannot add a shutdownHook

这个错误表明在 Spark 任务执行过程中，尝试向 Hadoop 的 ShutdownHookManager 注册一个关闭钩子（shutdown hook），但此时系统已经处于关闭状态。这通常是因为多个组件试图注册相同的关闭钩子，或者在 Spark 生命周期中过晚地注册了关闭钩子。

---

✅ 问题分析

1. 版本兼容性问题

• 你使用的版本是：

  • Spark 3.5.3
  • Atlas 2.4.0
  • Kyuubi 1.11.0

  这些版本之间可能存在兼容性问题。尤其是 Kyubbi 的 lineage 插件与 Atlas 的兼容性 可能存在问题，尤其是在 Spark 3.5.x 中的某些行为变化。

2. 重复注册 Shutdown Hook

• 在 Spark 应用程序中，每个组件都可能尝试注册自己的关闭钩子。
• 如果 Kyuubi 的 Lineage 插件 和 Hadoop 或 Spark 自身 都尝试注册相同的关闭钩子，就会导致冲突。
• 错误信息中的关键部分是：

  at org.apache.hadoop.util.ShutdownHookManager.addShutdownHook(ShutdownHookManager.java:301)
  

---

🔧 解决方案

✅ 解决方案一：升级或降级相关组件版本

1. 升级 Kyuubi 到更高版本

• 检查是否 Kyuubi 1.11.0 支持 Spark 3.5.3 和 Atlas 2.4.0。
• 建议尝试使用 Kyuubi 1.12.x 或更高版本，这些版本可能已经修复了与 Atlas 的兼容性问题。

示例：

# 使用 Kyuubi 1.12.0 或更新版本
kyuubi_version = "1.12.0"

2. 升级 Atlas 到兼容版本

• 确保 Atlas 版本与 Spark 兼容。
• Atlas 2.4.0 对应的 Spark 最佳支持版本为 Spark 3.3.x 左右，建议尝试 Spark 3.4.x。

---

✅ 解决方案二：禁用 Kyuubi Lineage 插件的 Shutdown Hook 注册

如果你暂时无法升级版本，可以尝试通过修改 Kyuubi 的源码来绕过该问题。

修改步骤如下：

1. 找到 Kyuubi 的 AtlasClient.scala 文件，路径类似：

   kyuubi-spark-lineage_2.12-1.11.0.jar!/org/apache/kyuubi/plugin/lineage/dispatcher/atlas/AtlasClient.scala
   

2. 定位到以下代码段（大约第88行）：

   registerCleanupShutdownHook()
   

3. 将其替换为：

   if (!isShuttingDown) {
     registerCleanupShutdownHook()
   }
   

   或者直接注释掉该部分，避免注册关闭钩子。

注意：此操作需要重新打包 Kyuubi 的插件包，适用于测试环境或临时解决方式。

---

✅ 解决方案三：配置 Spark 禁止自动注册 Shutdown Hook

你可以通过设置 Spark 的 JVM 参数来防止 Hadoop 注册关闭钩子，从而避免冲突。

添加如下 JVM 参数：

-Dhadoop.shutdown.hook.enabled=false

示例配置（如在 spark-submit 中）：

spark-submit \
  --conf "spark.driver.extraJavaOptions=-Dhadoop.shutdown.hook.enabled=false" \
  --conf "spark.executor.extraJavaOptions=-Dhadoop.shutdown.hook.enabled=false" \
  your_app.py

---

📌 总结：关键步骤

1. 检查并升级 Kyuubi 和 Atlas 的版本，确保兼容性。
2. 如果无法升级，可尝试禁用 Kyuubi 的 Shutdown Hook 注册逻辑。
3. 通过 JVM 参数禁止 Hadoop 注册关闭钩子。

---

📝 附录：修改后的代码示例（仅用于调试）

假设你找到了 AtlasClient.scala 并准备修改：

// 修改前
registerCleanupShutdownHook()

// 修改后（添加条件判断）
if (!isShuttingDown) {
  registerCleanupShutdownHook()
}

请确保你了解修改的影响，并在测试环境中验证后再部署生产环境。

---

❗️ 注意事项

• 该问题可能导致 Lineage 数据未正确发送到 Atlas，影响数据血缘追踪功能。
• 如果你无法升级版本，建议先在测试环境中验证上述修改是否有效。

如需进一步帮助，欢迎提供完整的日志和配置文件内容。