提交Spark应用至Amazon EMR时出现org.apache.spark.sql.catalyst.FileSourceOptions$类未找到异常

问题背景 #

应用配置（build.sbt） #

name := "IngestFromS3ToKafka"
version := "1.0"
scalaVersion := "2.12.17"
resolvers += "confluent" at "https://packages.confluent.io/maven/"

val sparkVersion = "3.3.1"

libraryDependencies ++= Seq(
  "org.apache.spark" %% "spark-core" % sparkVersion % "provided",
  "org.apache.spark" %% "spark-sql" % sparkVersion % "provided",
  "org.apache.hadoop" % "hadoop-common" % "3.3.5" % "provided",
  "org.apache.hadoop" % "hadoop-aws" % "3.3.5" % "provided",
  "com.amazonaws" % "aws-java-sdk-bundle" % "1.12.475" % "provided",

  "org.apache.spark" %% "spark-avro" % sparkVersion,
  "org.apache.spark" %% "spark-sql-kafka-0-10" % sparkVersion,
  "io.delta" %% "delta-core" % "2.4.0",
  "za.co.absa" %% "abris" % "6.3.0"
)

ThisBuild / assemblyMergeStrategy := {
  case PathList("module-info.class") => MergeStrategy.discard
  case x if x.endsWith("/module-info.class") => MergeStrategy.discard
  case PathList("org", "apache", "spark", "unused", "UnusedStubClass.class") => MergeStrategy.first
  case x =>
    val oldStrategy = (ThisBuild / assemblyMergeStrategy).value
    oldStrategy(x)
}

提交命令与错误信息 #

使用内置Spark 3.3.1的Amazon EMR集群，执行以下提交命令：

/bin/bash -c "/usr/bin/spark-submit --master yarn --deploy-mode client --class com.hongbomiao.IngestFromS3ToKafka --name ingest-from-s3-to-kafka /home/hadoop/IngestFromS3ToKafka-assembly-1.0.jar"

出现核心错误：

Exception in thread "main" com.google.common.util.concurrent.ExecutionError: java.lang.NoClassDefFoundError: org/apache/spark/sql/catalyst/FileSourceOptions$
...
Caused by: java.lang.ClassNotFoundException: org.apache.spark.sql.catalyst.FileSourceOptions$

问题原因 #

Delta Lake版本与Spark版本不兼容。当前使用的Delta Lake 2.4.0对应Spark 3.4.x版本，而集群中的Spark是3.3.1，org.apache.spark.sql.catalyst.FileSourceOptions$类是Spark 3.4新增的，因此Delta代码在Spark 3.3.1环境下找不到该类。

解决方法 #

将Delta Lake版本降级到与Spark 3.3.1兼容的2.3.0版本，同时添加provided标记避免与EMR预装依赖冲突：

修改build.sbt中的Delta依赖：

"io.delta" %% "delta-core" % "2.3.0" % "provided"

重新打包应用后，再次执行spark-submit命令即可。

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