为啥 AWS EMR 上的 Spark 不从应用程序 fat jar 加载类？

Posted 2023-04-17

【中文标题】为啥 AWS EMR 上的 Spark 不从应用程序 fat jar 加载类？

【英文标题】：Why Spark on AWS EMR doesn't load class from application fat jar?为什么 AWS EMR 上的 Spark 不从应用程序 fat jar 加载类？

【发布时间】：2017-05-12 10:15:01

【问题描述】：

我的 spark 应用程序无法在 AWS EMR 集群上运行。我注意到这是因为某些类是从 EMR 设置的路径加载的，而不是从我的应用程序 jar 加载的。例如

java.lang.NoSuchMethodError: org.apache.avro.Schema$Field.<init>(Ljava/lang/String;Lorg/apache/avro/Schema;Ljava/lang/String;Ljava/lang/Object;)V
        at com.sksamuel.avro4s.SchemaFor$.fieldBuilder(SchemaFor.scala:424)
        at com.sksamuel.avro4s.SchemaFor$.fieldBuilder(SchemaFor.scala:406)

这里 org.apache.avro.Schema 是从 "jar:file:/usr/lib/spark/jars/avro- 加载的1.7.7.jar!/org/apache/avro/Schema.class"

而 com.sksamuel.avro4s 依赖于 avro 1.8.1。我的应用程序是作为一个胖罐子构建的，并且有 avro 1.8.1。为什么没有加载？而不是从 EMR 设置类路径中选择 1.7.7。

这只是一个例子。我在我的应用程序中包含的其他库也看到了同样的情况。可能是 Spark 依赖于 1.7.7，在包含其他依赖项时我不得不遮蔽。但是为什么我的 app jar 中包含的类没有先加载呢？

【参考方案1】：

经过一番阅读，我意识到这就是 Spark 中类加载的工作方式。有一个钩子可以改变这种行为spark.executor.userClassPathFirst。当我尝试并将其标记为实验性时，它并没有很好地工作。我想最好的方法是遮蔽依赖关系。考虑到 Spark 及其组件拉取的库的数量，这对于复杂的 Spark 应用程序来说可能会有很多阴影。

【参考方案2】：

我和你有同样的例外。基于recommendation，我能够按照您的建议通过对 avro 依赖项进行着色来解决此异常：

assemblyShadeRules in assembly := Seq( ShadeRule.rename("org.apache.avro.**" -> "latest_avro.@1").inAll )

如果有帮助，这是我的完整 build.sbt（项目信息除外）：

val sparkVersion = "2.1.0"
val confluentVersion = "3.2.1"

resolvers += "Confluent" at "http://packages.confluent.io/maven"

libraryDependencies ++= Seq(
  "org.scala-lang" % "scala-library" % scalaVersion.value % "provided",
  "org.scala-lang" % "scala-reflect" % scalaVersion.value % "provided",
  "org.apache.spark" %% "spark-streaming" % sparkVersion % "provided",
  "org.apache.spark" % "spark-streaming-kafka-0-10_2.11" % sparkVersion,
  "org.apache.spark" %% "spark-sql" % sparkVersion % "provided" excludeAll ExclusionRule(organization = "org.scala-lang"),
  "org.apache.avro" % "avro" % "1.8.1" % "provided",
  "com.databricks" %% "spark-avro" % "3.2.0",
  "com.sksamuel.avro4s" %% "avro4s-core" % "1.6.4",
  "io.confluent" % "kafka-avro-serializer" % confluentVersion
)

logBuffered in Test := false

assemblyShadeRules in assembly := Seq(
  ShadeRule.rename("shapeless.**" -> "new_shapeless.@1").inAll,
  ShadeRule.rename("org.apache.avro.**" -> "latest_avro.@1").inAll
)

assemblyMergeStrategy in assembly := 
  case PathList("META-INF", xs @ _*) => MergeStrategy.discard
  case x =>
    val oldStrategy = (assemblyMergeStrategy in assembly).value
    oldStrategy(x)