如何使用 spark-streaming-kafka-0-8 依赖火花提交 Spark Streaming 应用程序？答案

【问题标题】：How to spark-submit a Spark Streaming application with spark-streaming-kafka-0-8 dependency?如何使用 spark-streaming-kafka-0-8 依赖火花提交 Spark Streaming 应用程序？
【发布时间】：2017-08-16 22:28:06
【问题描述】：

我正在尝试运行火花流示例： Directkafkawordcount.scala

要创建 jar，我使用带有插件的“build.sbt”：

     name := "Kafka Direct"

version := "1.0"

scalaVersion := "2.11.6"

libraryDependencies ++= Seq ("org.apache.spark" % "spark-streaming-kafka-0-8_2.11" % "2.1.0",
"org.apache.spark" % "spark-streaming_2.11" % "2.1.0",
"org.apache.spark" % "spark-core_2.11" % "2.1.0"  exclude("com.esotericsoftware.minlog", "minlog")
)
resolvers ++= Seq(
    "Maven Central" at "https://repo1.maven.org/maven2/"
)



 mergeStrategy in assembly := {
  case m if m.toLowerCase.endsWith("manifest.mf")          =>     MergeStrategy.discard
  case m if m.toLowerCase.matches("meta-inf.*\\.sf$")      =>     MergeStrategy.discard
  case "log4j.properties"                                  =>     MergeStrategy.discard
  case m if m.toLowerCase.startsWith("meta-inf/services/") =>     MergeStrategy.filterDistinctLines
  case "reference.conf"                                    =>     MergeStrategy.concat
  case _                                                   =>     MergeStrategy.first
  case PathList(ps @ _*) if ps.last endsWith "pom.properties" =>  MergeStrategy.discard
  case x => val oldStrategy = (assemblyMergeStrategy in assembly).value
  oldStrategy(x)
}

"sbt package" 成功但是当我想从 target/scala-2.11/classes 目录提交时： “spark-submit --class org.apache.spark.examples.streaming.DirectKafkaWordCount --master local[2] /home/hadoop/DirectKafkaProject/target/scala-2.11/kafka-direct_2.11-1.0.jar localhost:9092 xyz123"

它给了我这个错误：

引起：java.lang.ClassNotFoundException：org.apache.spark.streaming.kafka.KafkaUtils$ 在 java.net.URLClassLoader$1.run(URLClassLoader.java:366)

我已经设置了 SPARK_CLASSPATH 和 SPARK_LOCAL_IP。我已经尝试使用 -jars 选项，但它要求另一个 .jar 文件并且它一直在要求其他 .jar 文件。我已经按照这个网站的建议做了每一件事，但我无法解决我的问题。斯卡拉版本：2.11.6 火花版本：2.1.0 卡夫卡版本：2.11-0.10.2.0 .

请帮帮我。谢谢。

【问题讨论】：

您需要添加 Kafka jar、Kafa utils、metrics core jar 来触发类路径。
我的 spark 类路径是用 spark 的 jar 和 kafka jar 设置的。
我尝试使用这个“spark-submit --class org.apache.spark.examples.streaming.DirectKafkaWordCount --jars /home/hadoop/jars/spark-streaming-kafka-0-8_2。 11-2.1.0.jar,/home/hadoop/jars/kafka-clients-0.10.2.0.jar,/home/hadoop/jars/metrics-core-2.2.0.jar,/home/hadoop/jars/kafka_2 .11-0.10.2.0.jar --master local[2] /home/hadoop/DirectKafkaProject/target/scala-2.11/kafka-direct_2.11-1.0.jar localhost:9092 MyTest1" 。但现在它显示“线程中的异常”main“java.lang.ClassCastException：kafka.cluster.BrokerEndPoint 无法转换为 kafka.cluster.Broker”。所以，请帮助我。谢谢。
但最后一条命令显示“线程“main”中的异常 java.lang.ClassCastException：kafka.cluster.BrokerEndPoint 无法转换为 kafka.cluster.Broker”这个异常。谢谢。
您确定您机器上的 kafka 已启动并正在运行吗？并尝试将 spark-kafka-streaming jar 添加到 spark 和 kafka 类路径，然后重试。

标签： apache-spark apache-kafka spark-streaming

【解决方案1】：

tl;dr sbt assembly 和 spark-submit

sbt assembly 将所有依赖项组装在一个 jar 文件中，并且我建议您使用 Kafka 的外部库。

您已经在build.sbt 中获得了sbt-assembly 插件的定义，这表明这是要走的路。

执行组装的Spark应用程序使用spark-submit命令行应用程序如下：

bin/spark-submit \
  --class org.apache.spark.examples.streaming.DirectKafkaWordCount \
  --master local[2] \
  target/scala-2.11/StreamKafkaPr‌og-assembly-1.0.jar localhost:9092 rat123

【讨论】：

我以前和你一样想，但它对我不起作用。这就是为什么我使用 spark-submit 命令 --jars 选项。但它不起作用是版本不匹配或其他什么问题吗？
感谢亚切克·拉斯科瓦基！你说的对。一旦我完成了 sbt 组装，我的问题就解决了。在 build.sbt 文件中我做了一点改动： libraryDependencies ++= Seq( "org.apache.spark" % "spark-streaming_2.11" % "2.1.0" % "provided", "org.apache.spark " % "spark-streaming-kafka-0-8_2.11" % "2.1.0", "org.apache.spark" % "spark-core_2.11" % "2.1.0" % "提供的"解析器++ = Seq（“repo1.maven.org/maven2”处的“Maven Central”，“mvnrepository.com/artifact”处的“Kafka Central”）

【解决方案2】：

以防万一有人偶然发现这篇文章（就像我一样）。 Jacek 的回答非常正确。对我来说它仍然不起作用，因为合并策略至关重要（我使用了一些来自互联网的示例）。对我有用的最简单的形式（假设你只有spark 和spark-sql-kafka 等）是

assemblyMergeStrategy in assembly := {
  case PathList("org", "apache", "spark", "unused", "UnusedStubClass.class") => MergeStrategy.first
  case x => (assemblyMergeStrategy in assembly).value(x)
}

【讨论】：