解决Apache Spark中的依赖性问题

Question

构建和部署Spark应用程序时的常见问题是：

• java.lang.ClassNotFoundException。
• [object x is not a member of package y编译错误。
• java.lang.NoSuchMethodError

如何解决这些问题？

Answer 1

Apache Spark的类路径是动态构建的（以适应每个应用程序的用户代码），这使其容易受到此类问题的影响。 @user7337271的答案是正确的，但是还有更多的问题，这取决于您使用的集群管理器（“主服务器”）。

首先，Spark应用程序由这些组件组成（每个组件都是一个单独的JVM，因此可能在其类路径中包含不同的类）：

1. Driver：这是your应用程序，它创建一个SparkSession（或SparkContext）并连接到集群管理器以执行实际的工作
2. Cluster Manager：充当群集的“入口点”，负责为每个应用程序分配executors。 Spark支持几种不同的类型：独立，YARN和Mesos，我们将在下面介绍。
3. Executors
：这些是集群节点上的进程，它们在执行实际工作（运行Spark tasks）

这些关系之间的关系在此图中来自Apache Spark的cluster mode overview：

现在-哪些类应该驻留在每个组件中？

这可以通过下图来回答：

让我们慢慢解析：

1. Spark Code

   是Spark的库。它们应该存在于[[ALL三个组件中，因为它们包含让我们Spark在它们之间进行通信的胶水。顺便说一下，Spark作者做出了一个设计决策，即在所有组件中都包含所有组件的代码（例如，也应只在驱动程序的Executor中运行的代码）以简化此操作-因此，Spark的“胖罐”（版本不超过1.6） ）或“归档”（在2.0中，详细信息如下）包含所有组件的必要代码，并且在所有组件中都应可用。

2. 仅驱动程序代码

这是用户代码，其中不包含应在执行程序上使用的任何内容，即RDD / DataFrame / Dataset上的任何转换均未使用的代码。这不必一定要与分布式用户代码分开，但可以这样。

3. Distributed Code

这是使用驱动程序代码编译的用户代码，但也必须在执行程序上执行-实际转换所使用的所有内容都必须包含在此jar中。 现在我们明白了，

如何

是否让类正确地加载到每个组件中，它们应该遵循哪些规则？

1. Spark Code

   ：如先前的回答所述，您必须在所有组件中使用相同的Scala和Spark版本。

1.1在

Standalone

模式下，存在一个“现有” Spark安装，应用程序（驱动程序）可以连接到该安装。这意味着所有驱动程序必须使用相同的Spark版本在主服务器和执行程序上运行。1.2在

YARN / Mesos

中，每个应用程序可以使用不同的Spark版本，但是同一应用程序的所有组件必须使用相同的版本。这意味着，如果您使用版本X来编译和打包驱动程序应用程序，则在启动SparkSession时应提供相同的版本（例如，使用YARN时通过spark.yarn.archive或spark.yarn.jars参数）。您提供的jar /档案文件应包含所有Spark依赖项（包括传递性依赖项>>），并且在应用程序启动时，它将由集群管理器运送到每个执行程序。

驱动程序代码

：完全取决于-驱动程序代码可以一堆jar或“胖罐”的形式运送，只要它包含所有Spark依赖项和所有用户代码

]

Distributed Code

：除了存在于驱动程序上之外，此代码还必须交付给执行者（同样，连同其所有传递依赖关系一起）。这是使用spark.jars参数完成的。

总结

，这是构建和部署Spark应用程序的建议方法（在这种情况下-使用YARN）：

用您的分布式代码创建一个库，将其打包为“常规” jar（带有描述其依赖性的.pom文件）和“ fat jar”（包括其所有传递性依赖性）。
• 创建一个驱动程序应用程序，在分布式代码库和Apache Spark（具有特定版本）上具有编译依赖性。>>
• 将驱动程序应用程序打包到一个胖罐中，以部署到驱动程序中

启动spark.jars时，将正确的分布式代码版本作为SparkSession参数的值传递>

将下载的Spark二进制文件的lib/文件夹下包含所有jar的存档文件（例如gzip）的位置作为spark.yarn.archive的值传递

构建和部署Spark应用程序时，所有依赖项都需要兼容的版本。

• Scala版本

。所有软件包都必须使用相同的主要（2.10、2.11、2.12）Scala版本。

考虑以下（错误的）build.sbt：

name := "Simple Project"

version := "1.0"

libraryDependencies ++= Seq(
   "org.apache.spark" % "spark-core_2.11" % "2.0.1",
   "org.apache.spark" % "spark-streaming_2.10" % "2.0.1",
   "org.apache.bahir" % "spark-streaming-twitter_2.11" % "2.0.1"
)

我们为Scala 2.10使用spark-streaming，而其余软件包则为Scala 2.11。 
有效
文件可能是

name := "Simple Project"

version := "1.0"

libraryDependencies ++= Seq(
   "org.apache.spark" % "spark-core_2.11" % "2.0.1",
   "org.apache.spark" % "spark-streaming_2.11" % "2.0.1",
   "org.apache.bahir" % "spark-streaming-twitter_2.11" % "2.0.1"
)

但是最好是全局指定版本并使用%%（为您添加scala版本）：
name := "Simple Project"

version := "1.0"

scalaVersion := "2.11.7"

libraryDependencies ++= Seq(
   "org.apache.spark" %% "spark-core" % "2.0.1",
   "org.apache.spark" %% "spark-streaming" % "2.0.1",
   "org.apache.bahir" %% "spark-streaming-twitter" % "2.0.1"
)

类似地在Maven：

    <project>
      <groupId>com.example</groupId>
      <artifactId>simple-project</artifactId>
      <modelVersion>4.0.0</modelVersion>
      <name>Simple Project</name>
      <packaging>jar</packaging>
      <version>1.0</version>
      <properties>
        <spark.version>2.0.1</spark.version>
      </properties> 
      <dependencies>
        <dependency> <!-- Spark dependency -->
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-core_2.11</artifactId>
          <version>${spark.version}</version>
        </dependency>
        <dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-streaming_2.11</artifactId>
          <version>${spark.version}</version>
        </dependency> 
        <dependency>
          <groupId>org.apache.bahir</groupId>
          <artifactId>spark-streaming-twitter_2.11</artifactId>
          <version>${spark.version}</version>
        </dependency>
      </dependencies>
    </project>

• Spark version所有程序包都必须使用相同的主要Spark版本（1.6、2.0、2.1等）。

考虑以下（

不正确

）build.sbt：

name := "Simple Project"

version := "1.0"

libraryDependencies ++= Seq(
   "org.apache.spark" % "spark-core_2.11" % "1.6.1",
   "org.apache.spark" % "spark-streaming_2.10" % "2.0.1",
   "org.apache.bahir" % "spark-streaming-twitter_2.11" % "2.0.1"
)

我们使用spark-core 1.6，而其余组件位于Spark 2.0中。 
有效

文件可能是

name := "Simple Project"

version := "1.0"

libraryDependencies ++= Seq(
   "org.apache.spark" % "spark-core_2.11" % "2.0.1",
   "org.apache.spark" % "spark-streaming_2.10" % "2.0.1",
   "org.apache.bahir" % "spark-streaming-twitter_2.11" % "2.0.1"
)

但是最好
使用变量
（仍然不正确）：

name := "Simple Project"

version := "1.0"

val sparkVersion = "2.0.1"

libraryDependencies ++= Seq(
   "org.apache.spark" % "spark-core_2.11" % sparkVersion,
   "org.apache.spark" % "spark-streaming_2.10" % sparkVersion,
   "org.apache.bahir" % "spark-streaming-twitter_2.11" % sparkVersion
)

类似地在Maven：
    <project>
      <groupId>com.example</groupId>
      <artifactId>simple-project</artifactId>
      <modelVersion>4.0.0</modelVersion>
      <name>Simple Project</name>
      <packaging>jar</packaging>
      <version>1.0</version>
      <properties>
        <spark.version>2.0.1</spark.version>
        <scala.version>2.11</scala.version>
      </properties> 
      <dependencies>
        <dependency> <!-- Spark dependency -->
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-core_${scala.version}</artifactId>
          <version>${spark.version}</version>
        </dependency>
        <dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-streaming_${scala.version}</artifactId>
          <version>${spark.version}</version>
        </dependency> 
        <dependency>
          <groupId>org.apache.bahir</groupId>
          <artifactId>spark-streaming-twitter_${scala.version}</artifactId>
          <version>${spark.version}</version>
        </dependency>
      </dependencies>
    </project>
Spark依赖项中使用的Spark版本必须与Spark安装的Spark版本匹配。例如
如果在群集上使用1.6.1，则必须使用1.6.1来构建jars

。次版本不匹配并不总是可以接受。

用于构建jar的Scala版本必须与用于构建已部署Spark的Scala版本匹配。默认情况下（可下载的二进制文件和默认版本）：

Spark 1.x-> Scala 2.10
• Spark 2.x-> Scala 2.11

如果胖子罐中包含其他软件包，则应该在工作节点上可以访问其他软件包。有很多选项，包括：

--jars的spark-submit参数-分发本地jar文件。

--packages的参数[-spark-submit-从Maven存储库中获取依赖项。

在集群节点中提交时，应在jar中包括应用程序--jars。

除了user7337271已经给出的非常广泛的答案之外，如果问题是由于缺少外部依赖项而引起的，您可以使用例如maven assembly plugin

在这种情况下，请确保在构建系统中将所有核心Spark依赖项标记为“已提供”，并且如前所述，确保它们与您的运行时Spark版本相关。

您的应用程序的依赖关系类应在启动命令的

application-jar

选项中指定。 更多详细信息，请参见Spark documentation

摘自文档：

application-jar：包含您的应用程序和所有依赖项。该网址必须在您的内部全局可见群集，例如，hdfs：//路径或file：//路径存在于所有节点上

我认为这个问题必须解决一个程序集插件。您需要建立一个胖子罐。例如在sbt：

添加代码为$PROJECT_ROOT/project/assembly.sbt的文件addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.0")
• 至build.sbtadded some librarieslibraryDependencies ++ = Seq（“ com.some.company” %%“ some-lib”％“ 1.0.0”）`
• 在sbt控制台中输入“程序集”，并部署程序集jar

如果需要更多信息，请转到https://github.com/sbt/sbt-assembly

从项目中的spark-2.4.0-bin-hadoop2.7 \ spark-2.4.0-bin-hadoop2.7 \ jars中添加所有jar文件。可以从https://spark.apache.org/downloads.html]下载spark-2.4.0-bin-hadoop2.7

我有以下build.sbt

lazy val root = (project in file(".")).
  settings(
    name := "spark-samples",
    version := "1.0",
    scalaVersion := "2.11.12",
    mainClass in Compile := Some("StreamingExample")        
  )

libraryDependencies ++= Seq(
  "org.apache.spark" %% "spark-core" % "2.4.0",
  "org.apache.spark" %% "spark-streaming" % "2.4.0",
  "org.apache.spark" %% "spark-sql" % "2.4.0",
  "com.couchbase.client" %% "spark-connector" % "2.2.0" 
)

// META-INF discarding
assemblyMergeStrategy in assembly := {
       case PathList("META-INF", xs @ _*) => MergeStrategy.discard
       case x => MergeStrategy.first
   }

我已经使用sbt程序集插件为我的应用创建了一个胖子，但是使用spark-submit运行时失败，并显示以下错误：

java.lang.NoClassDefFoundError: rx/Completable$OnSubscribe
    at com.couchbase.spark.connection.CouchbaseConnection.streamClient(CouchbaseConnection.scala:154)

我可以看到该类存在于我的胖子罐中：

jar tf target/scala-2.11/spark-samples-assembly-1.0.jar | grep 'Completable$OnSubscribe'
rx/Completable$OnSubscribe.class

不确定我在这里想念什么，有什么线索吗？ 

Answer 2

构建和部署Spark应用程序时，所有依赖项都需要兼容的版本。

• Scala版本

。所有软件包都必须使用相同的主要（2.10、2.11、2.12）Scala版本。

考虑以下（错误的）build.sbt：

Answer 3

除了user7337271已经给出的非常广泛的答案之外，如果问题是由于缺少外部依赖项而引起的，您可以使用例如maven assembly plugin

在这种情况下，请确保在构建系统中将所有核心Spark依赖项标记为“已提供”，并且如前所述，确保它们与您的运行时Spark版本相关。

Answer 4

您的应用程序的依赖关系类应在启动命令的

application-jar

Answer 5

我认为这个问题必须解决一个程序集插件。您需要建立一个胖子罐。例如在sbt：

添加代码为$PROJECT_ROOT/project/assembly.sbt的文件addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.0")
• 至build.sbtadded some librarieslibraryDependencies ++ = Seq（“ com.some.company” %%“ some-lib”％“ 1.0.0”）`
• 在sbt控制台中输入“程序集”，并部署程序集jar

Answer 6

从项目中的spark-2.4.0-bin-hadoop2.7 \ spark-2.4.0-bin-hadoop2.7 \ jars中添加所有jar文件。可以从https://spark.apache.org/downloads.html]下载spark-2.4.0-bin-hadoop2.7

我有以下build.sbt

lazy val root = (project in file(".")).
  settings(
    name := "spark-samples",
    version := "1.0",
    scalaVersion := "2.11.12",
    mainClass in Compile := Some("StreamingExample")        
  )

libraryDependencies ++= Seq(
  "org.apache.spark" %% "spark-core" % "2.4.0",
  "org.apache.spark" %% "spark-streaming" % "2.4.0",
  "org.apache.spark" %% "spark-sql" % "2.4.0",
  "com.couchbase.client" %% "spark-connector" % "2.2.0" 
)

// META-INF discarding
assemblyMergeStrategy in assembly := {
       case PathList("META-INF", xs @ _*) => MergeStrategy.discard
       case x => MergeStrategy.first
   }

我已经使用sbt程序集插件为我的应用创建了一个胖子，但是使用spark-submit运行时失败，并显示以下错误：

java.lang.NoClassDefFoundError: rx/Completable$OnSubscribe
    at com.couchbase.spark.connection.CouchbaseConnection.streamClient(CouchbaseConnection.scala:154)

我可以看到该类存在于我的胖子罐中：

Answer 7

我有以下build.sbt

lazy val root = (project in file(".")).
  settings(
    name := "spark-samples",
    version := "1.0",
    scalaVersion := "2.11.12",
    mainClass in Compile := Some("StreamingExample")        
  )

libraryDependencies ++= Seq(
  "org.apache.spark" %% "spark-core" % "2.4.0",
  "org.apache.spark" %% "spark-streaming" % "2.4.0",
  "org.apache.spark" %% "spark-sql" % "2.4.0",
  "com.couchbase.client" %% "spark-connector" % "2.2.0" 
)

// META-INF discarding
assemblyMergeStrategy in assembly := {
       case PathList("META-INF", xs @ _*) => MergeStrategy.discard
       case x => MergeStrategy.first
   }