java - 运行实例化 NiFi 接收器的 Spark 应用程序时出现依赖项错误

Question

一段时间以来，我一直在尝试解决这个问题，但似乎找不到答案。我正在用 Scala 编写一个简单的 Spark 应用程序，它实例化了一个 NiFi 接收器，虽然它使用 SBT 成功构建，但当我尝试使用 spark-submit 运行该应用程序时，我收到以下错误:

Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/nifi/spark/NiFiReceiver
    at <app name>.main(main.scala)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
    at java.lang.reflect.Method.invoke(Unknown Source)
    at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:731)
    at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:181)
    at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:206)
    at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:121)
    at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala) 
Caused by: java.lang.ClassNotFoundException: org.apache.nifi.spark.NiFiReceiver
        at java.net.URLClassLoader.findClass(Unknown Source)
        at java.lang.ClassLoader.loadClass(Unknown Source)
        at java.lang.ClassLoader.loadClass(Unknown Source)
        ... 10 more

我尝试了一些变体，但这是我的 build.sbt 文件:

name := "<application name here>"
version := "1.0"
scalaVersion := "2.10.5"

libraryDependencies += "org.apache.spark" %% "spark-core" % "1.6.2" % "provided"
libraryDependencies += "org.apache.spark" %% "spark-streaming" % "1.6.2" % "provided"
libraryDependencies += "org.apache.nifi" % "nifi-spark-receiver" % "0.7.0"
libraryDependencies += "org.apache.nifi" % "nifi-site-to-site-client" % "0.7.0"

应该注意的是，如果我将两行 nifi 更改为使用 Scala 等效项(即每行中的第一个百分号替换为两个百分号)，我实际上会在运行“sbt package”时收到以下错误:

[error] (*:update) sbt.ResolveException: unresolved dependency: org.apache.nifi#nifi-spark-receiver_2.10;0.7.0: not found
[error] unresolved dependency: org.apache.nifi#nifi-site-to-site-client_2.10;0.7.0: not found

正如我之前提到的，使用单个百分号(因此使用 Java 依赖项)我在构建时不会出错，但在运行时会出错。

我的应用程序的相关部分(删除了某些名称)如下:

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import java.time
import java.time._
import org.apache.nifi._
import java.nio.charset._
import org.apache.nifi.spark._
import org.apache.nifi.remote.client._
import org.apache.spark._
import org.apache.nifi.events._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._
import org.apache.nifi.remote._
import org.apache.nifi.remote.protocol._
import org.apache.spark.streaming.receiver._
import org.apache.spark.storage._
import java.io._
import org.apache.spark.serializer._

object <app name> {
    def main(args: Array[String]) {

        val nifiUrl = "<nifi url>"
        val nifiReceiverConfig = new SiteToSiteClient.Builder()
            .url(nifiUrl)
            .portName("Data for Spark")
            .buildConfig()

        val conf = new SparkConf().setAppName("<app name>")
        val ssc = new StreamingContext(conf, Seconds(10))
        val packetStream = ssc.receiverStream(new NiFiReceiver(nifiReceiverConfig, StorageLevel.MEMORY_ONLY))

错误指的是此处的最后一行，其中实例化了 NifiReceiver - 它似乎无法在任何地方找到该类名。

我已经尝试了很多方法，包括以下(分别): 1) 找到 nifi-spark-receiver 和 nifi-site-to-site-client 的 jar 文件并将它们添加到我项目的 lib 目录中 2)关注这篇文章https://community.hortonworks.com/articles/12708/nifi-feeding-data-to-spark-streaming.html .我最终在我的 Spark conf 目录中制作了 spark-default.conf.template 的副本，将其重命名为 spark-defaults.conf 并将步骤 1 中该链接处的两行添加到文件中(代替实际名称和位置有问题的文件)。然后我确保我拥有该页面上两个代码示例中使用的所有必要的导入语句 3) 在我的应用程序目录的根目录下创建一个项目目录，然后在其中创建一个名为assembly.sbt 的文件。然后我在里面添加了以下行(如此处引用:https://github.com/sbt/sbt-assembly):

addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.3")

然后我运行“sbt assembly”而不是“sbt package”来让应用程序创建一个 super jar，但是这也失败了，并出现了与在build.sbt 文件:

[error] (*:update) sbt.ResolveException: unresolved dependency: org.apache.nifi#nifi-spark-receiver_2.10;0.7.0: not found
[error] unresolved dependency: org.apache.nifi#nifi-site-to-site-client_2.10;0.7.0: not found

如果需要任何进一步的信息，请告诉我。在此先感谢您的帮助。

Answer 1

好的，所以我已经设法解决了这个问题，这里是任何可能感兴趣的人的答案:

答案是回到 uber-jar 路线并使用“sbt assembly”而不是“sbt package”来在我的 uber-jar 中包含必要的依赖 jar。

1) 在根目录下创建一个名为“project”的目录，并在其中放置一个名为 assembly.sbt 的文件，其中包含以下内容(我最初尝试在此处添加的是 resolvers 行):

resolvers += Resolver.url("sbt-plugin-releases-scalasbt", url("http://repo.scala-sbt.org/scalasbt/sbt-plugin-releases/"))

addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.3")

2) 在项目根目录的 build.sbt 文件中，使用以下依赖引用(即不是特定于 spark 版本):

libraryDependencies += "org.apache.nifi" % "nifi-spark-receiver" % "0.7.0"
libraryDependencies += "org.apache.nifi" % "nifi-site-to-site-client" % "0.7.0"

我还将 spark-core 和 spark-streaming 标记为“已提供”，即

libraryDependencies += "org.apache.spark" %% "spark-core" % "1.6.2" % "provided"
libraryDependencies += "org.apache.spark" %% "spark-streaming" % "1.6.2" % "provided"

这意味着您需要单独提供 spark，但它不会让 uber-jar 变得更大。

3) 在同一个文件中，添加如下代码来处理拉取依赖时的合并(这很重要):

mergeStrategy in assembly <<= (mergeStrategy in assembly) { (old) =>
   {
    case PathList("META-INF", xs @ _*) => MergeStrategy.discard
    case x => MergeStrategy.first
   }
}

4) 确保相关的导入语句存在于你的 scala 文件中，例如

import org.apache.nifi._
import org.apache.nifi.spark._

等等

然后，当您运行“sbt assembly”时，它应该会成功构建——只需在调用“spark-submit”时引用这个 jar，即

spark-submit --class "<application class name>" --master "<local or url>" "<path to uber-jar from project root directory>"

请注意，以下帖子对找到此解决方案提供了巨大帮助:java.lang.NoClassDefFoundError: org/apache/spark/streaming/twitter/TwitterUtils$ while running TwitterPopularTags