Spark-Submit提交Kafka流任务报ClassNotFoundException问题求助

我来帮你拆解这个问题的核心原因和可行的解决办法——你遇到的报错本质是Spark无法找到Kafka数据源的实现类，虽然本地IDE能正常运行，但打包和集群提交时的依赖处理逻辑和本地环境完全不同，才导致了这个冲突。

问题根源分析 #

1. Assembly打包的合并策略问题：你当前的assemblyMergeStrategy丢弃了大部分META-INF目录下的文件，而Spark的数据源注册正是通过META-INF/services/org.apache.spark.sql.sources.DataSourceRegister这个文件来定位kafka.DefaultSource的。虽然你对这个文件设置了concat，但其他关键的依赖元信息被丢弃，导致打包后的jar里缺失了数据源注册的必要信息。
2. 提交命令的依赖冲突：你提交时同时使用了assembly打包的胖包和--packages参数，两者的依赖加载逻辑会发生冲突——Spark优先使用胖包内的依赖，但--packages的引入反而打乱了依赖的优先级，导致关键类无法被正确加载。

解决办法（按优先级推荐） #

方案一：调整Assembly打包策略，正确包含Kafka依赖

修改你的build.sbt，优化依赖配置和合并规则：

1. 保留Spark核心依赖的provided标记（避免和集群Spark版本冲突），仅将Kafka相关依赖改为编译依赖：

libraryDependencies ++= Seq(
  "org.apache.spark" %% "spark-core" % "2.4.0" % "provided",
  "org.apache.spark" %% "spark-sql" % "2.4.0" % "provided",
  "org.apache.spark" %% "spark-streaming" % "2.4.0" % "provided",
  // 移除这两个Kafka依赖的provided标记，确保被打包进jar
  "org.apache.spark" %% "spark-streaming-kafka-0-10" % "2.4.0",
  "org.apache.spark" %% "spark-sql-kafka-0-10" % "2.4.0"
)

2. 优化合并策略，保留必要的元信息文件：

assemblyMergeStrategy in assembly := {
  // 合并所有数据源注册文件，确保Kafka数据源能被识别
  case "META-INF/services/org.apache.spark.sql.sources.DataSourceRegister" => MergeStrategy.concat
  // 保留签名文件，避免包验证失败
  case "META-INF/*.SF" | "META-INF/*.DSA" | "META-INF/*.RSA" => MergeStrategy.first
  // 仅丢弃maven相关的冗余元信息，不要全部丢弃META-INF
  case PathList("META-INF", "maven", _*) => MergeStrategy.discard
  case PathList("META-INF", _*) => MergeStrategy.first
  case _ => MergeStrategy.first
}

3. 重新执行sbt assembly打包，提交时不要加--packages参数（依赖已打包进jar）：

spark-submit --class com.ibm.kafkasparkintegration.executables.WeatherDataStream hdfs://<some address>:8020/user/clsadmin/consumer-example.jar

方案二：使用瘦包+--packages提交（更轻量）

如果你不想打包胖包，可以恢复依赖的provided标记，用普通打包+动态拉取依赖的方式：

1. 恢复build.sbt的provided标记：

libraryDependencies ++= Seq(
  "org.apache.spark" %% "spark-core" % "2.4.0" % "provided",
  "org.apache.spark" %% "spark-sql" % "2.4.0" % "provided",
  "org.apache.spark" %% "spark-streaming" % "2.4.0" % "provided",
  "org.apache.spark" %% "spark-streaming-kafka-0-10" % "2.4.0" % "provided",
  "org.apache.spark" %% "spark-sql-kafka-0-10" % "2.4.0" % "provided"
)

2. 执行sbt package生成瘦包，提交时仅指定Kafka相关依赖（注意匹配Scala版本）：

spark-submit --class com.ibm.kafkasparkintegration.executables.WeatherDataStream \
--packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.0 \
hdfs://<some address>:8020/user/clsadmin/consumer-example.jar

不需要指定所有Spark依赖，集群环境已经包含核心依赖。

方案三：集群全局配置（适合批量任务）

如果集群内多个任务都需要使用Kafka数据源，可以将spark-sql-kafka-0-10_2.11-2.4.0.jar复制到Spark集群的$SPARK_HOME/jars目录下，重启Spark服务后，所有任务都能直接使用该数据源。

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