AWS上でスパークする理由EMRはアプリケーションのfat jarからクラスをロードしませんか？

Question

私のsparkアプリケーションはAWS EMRクラスタ上で動作しません。私は、これは、いくつかのクラスが私のアプリケーションjarではなく、EMRによって設定されたパスからロードされていることに気づきました。ファイル：：！/usr/lib/spark/jars/avro-1.7.7.jar/orgの例ここで

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ファイルからロードされます/apache/avro/Schema.class "

一方、com.sksamuel.avro4sはavro 1.8.1に依存します。私のアプリケーションはfat jarとして構築され、avro 1.8.1を持っています。それはなぜロードされていないのですか？ EMRから1.7.7を選択する代わりにクラスパスを設定します。

これは単なる例です。私は私のアプリケーションに含まれている他のライブラリと同じことが分かります。 Sparkは1.7.7に依存しているので、他の依存関係を含めるときには陰にする必要があります。しかし、なぜ私のアプリjarに含まれるクラスが最初に読み込まれないのですか？

Answer 1

読書の後、私はこれがクラスロードがスパークでどのように動作するのかを理解しました。この動作を変更するにはフックがありますspark.executor.userClassPathFirst。試してみるとうまくいかず、実験的にマークされていました。私が進める最良の方法は、依存関係を陰にすることだと思います。 Sparkとそのコンポーネントが引き出すライブラリの数を考えると、これは複雑なSparkアプリケーションではかなり濃くなるかもしれません。

Answer 2

私はあなたと同じ例外を抱えていました。 recommendationに基づいて、私はあなたが示唆したようにアブロ依存関係を遮光することにより、この例外を解決することができた：

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) 
}