Spark3.0.3+Scala2.12.10从EventHub流数据到Delta Table报错求助

环境信息 #

• Spark 3.0.3
• Scala 2.12.10
• 数据源：Azure EventHub
• 目标存储：Delta Table

执行代码 #

devicesDF.writeStream.format("delta").outputMode("append").option("checkpointLocation", "/delta/events/table").start("/delta/table")

错误日志 #

22/09/24 11:47:46 ERROR MicroBatchExecution: Query [id = 651c88cf-0098-42b5-858d-a32ee9fecfdf, runId = 96fb3c81-79c3-492f-9be4-19235651a6a3] terminated with error
java.lang.NoClassDefFoundError: scala/compat/java8/FutureConverters$
    at org.apache.spark.eventhubs.utils.RetryUtils$.$anonfun$retryJava$1(RetryUtils.scala:91)
    at org.apache.spark.eventhubs.utils.RetryUtils$.org$apache$spark$eventhubs$utils$RetryUtils$$retryHelper$1(RetryUtils.scala:116)
    at org.apache.spark.eventhubs.utils.RetryUtils$.retryScala(RetryUtils.scala:149)
    at org.apache.spark.eventhubs.utils.RetryUtils$.retryJava(RetryUtils.scala:91)
    at org.apache.spark.eventhubs.client.ClientConnectionPool.org$apache$spark$eventhubs$client$ClientConnectionPool$$borrowClient(ClientConnectionPool.scala:83)
    at org.apache.spark.eventhubs.client.ClientConnectionPool$.borrowClient(ClientConnectionPool.scala:170)
    at org.apache.spark.eventhubs.client.EventHubsClient.client(EventHubsClient.scala:62)
    at org.apache.spark.eventhubs.client.EventHubsClient.liftedTree1$1(EventHubsClient.scala:187)
    at org.apache.spark.eventhubs.client.EventHubsClient.partitionCountLazyVal$lzycompute(EventHubsClient.scala:184)
    at org.apache.spark.eventhubs.client.EventHubsClient.partitionCountLazyVal(EventHubsClient.scala:183)
    at org.apache.spark.eventhubs.client.EventHubsClient.partitionCount(EventHubsClient.scala:176)
    at org.apache.spark.sql.eventhubs.EventHubsSource.partitionCount(EventHubsSource.scala:81)
    at org.apache.spark.sql.eventhubs.EventHubsSource.$anonfun$maxOffsetsPerTrigger$4(EventHubsSource.scala:96)
    at scala.runtime.java8.JFunction0$mcJ$sp.apply(JFunction0$mcJ$sp.java:23)
    at scala.Option.getOrElse(Option.scala:189)
    at org.apache.spark.sql.eventhubs.EventHubsSource.$anonfun$maxOffsetsPerTrigger$2(EventHubsSource.scala:96)
    at scala.runtime.java8.JFunction0$mcJ$sp.apply(JFunction0$mcJ$sp.java:23)
    at scala.Option.getOrElse(Option.scala:189)
    at org.apache.spark.sql.eventhubs.EventHubsSource.<init>(EventHubsSource.scala:96)
    at org.apache.spark.sql.eventhubs.EventHubsSourceProvider.createSource(EventHubsSourceProvider.scala:84)
    at org.apache.spark.sql.execution.datasources.DataSource.createSource(DataSource.scala:296)
    at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$1.$anonfun$applyOrElse$1(MicroBatchExecution.scala:86)
    at scala.collection.mutable.HashMap.getOrElseUpdate(HashMap.scala:86)
    at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$1.applyOrElse(MicroBatchExecution.scala:83)
    at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$1.applyOrElse(MicroBatchExecution.scala:81)
    at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDown$1(TreeNode.scala:317)
    at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:75)
    at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:317)
    at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDown(LogicalPlan.scala:29)
    at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDown(AnalysisHelper.scala:162)
    at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDown$(AnalysisHelper.scala:160)
    at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDown(LogicalPlan.scala:29)
    at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDown(LogicalPlan.scala:29)
    at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDown$3(TreeNode.scala:322)
    at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$mapChildren$1(TreeNode.scala:407)
    at org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:245)
    at org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:405)
    at org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:358)
    at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:322)
    at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDown(LogicalPlan.scala:29)
    at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDown(AnalysisHelper.scala:162)
    at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDown$(AnalysisHelper.scala:160)
    at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDown(LogicalPlan.scala:29)
    at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDown(LogicalPlan.scala:29)
    at org.apache.spark.sql.catalyst.trees.TreeNode.transform(TreeNode.scala:306)
    at org.apache.spark.sql.execution.streaming.MicroBatchExecution.logicalPlan$lzycompute(MicroBatchExecution.scala:81)
    at org.apache.spark.sql.execution.streaming.MicroBatchExecution.logicalPlan(MicroBatchExecution.scala:61)
    at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:322)
    at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:245)
Caused by: java.lang.ClassNotFoundException: scala.compat.java8.FutureConverters$
    at java.net.URLClassLoader.findClass(URLClassLoader.java:387)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:355)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
    ... 49 more

问题原因 #

1. 依赖缺失：scala.compat.java8.FutureConverters$是Scala 2.12用于兼容Java8异步API的标准组件，当前作业的classpath中缺少对应的scala-java8-compat库。
2. 版本不兼容：使用的Azure EventHub连接器版本与Spark 3.0.3、Scala 2.12.10不匹配，导致依赖冲突或间接缺失必要组件。

解决方案 #

1. 补充Scala Java8兼容库 #

根据构建工具添加对应依赖：

• Maven：

<dependency>
    <groupId>org.scala-lang.modules</groupId>
    <artifactId>scala-java8-compat_2.12</artifactId>
    <version>0.9.1</version>
</dependency>

• SBT：

libraryDependencies += "org.scala-lang.modules" %% "scala-java8-compat" % "0.9.1"

• Spark Submit命令：

spark-submit --packages org.scala-lang.modules:scala-java8-compat_2.12:0.9.1 [你的作业参数]

2. 匹配EventHub连接器版本 #

Spark 3.0.3对应兼容的EventHub连接器版本为2.3.22（适配Scala 2.12），在Spark Submit中同时指定：

spark-submit --packages com.microsoft.azure:azure-eventhubs-spark_2.12:2.3.22,org.scala-lang.modules:scala-java8-compat_2.12:0.9.1 [你的作业参数]

3. 确保Delta Lake版本兼容 #

Spark 3.0.3建议搭配Delta Lake 0.8.x版本，完整依赖命令：

spark-submit --packages io.delta:delta-core_2.12:0.8.0,com.microsoft.azure:azure-eventhubs-spark_2.12:2.3.22,org.scala-lang.modules:scala-java8-compat_2.12:0.9.1 [你的作业参数]

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