Scala SBT 程序集 kafka 流错误

Question

我是 Scala 和 SBT 的新手。我正在使用 Kafka 流式传输并将数据存储到 Cassandra DB。在尝试使用 sbt 汇编命令获取 fat jar 时，我遇到了下面提到的错误。

如何解决这个问题？并采取脂肪罐

build.sbt

organization := "com.example"
name := "cass-conn"
version := "0.1"
scalaVersion := "2.11.8"
val sparkVersion = "2.2.0"
val connectorVersion = "2.0.7"
val kafka_stream_version = "1.6.3"

libraryDependencies ++= Seq(
  "org.apache.spark" %% "spark-core" % sparkVersion  % "provided",
  "org.apache.spark" %% "spark-sql" % sparkVersion  % "provided",
  "org.apache.spark" %% "spark-hive" % sparkVersion  % "provided",
  "com.datastax.spark" %% "spark-cassandra-connector" % connectorVersion  ,
  "org.apache.spark" %% "spark-streaming-kafka-0-10" % "2.2.0",
  "org.apache.spark" %% "spark-streaming" %  "2.2.0"  % "provided",
)

plugins.sbt

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

SBT 版本：1.0.3

错误

[error] 1 error was encountered during merge
[error] java.lang.RuntimeException: deduplicate: different file contents found in the following:
[error] C:\Users\gnana\.ivy2\cache\org.apache.spark\spark-streaming-kafka-0-10_2.11\jars\spark-streaming-kafka-0-10_2.11-2.2.0.jar:org/apache/spark/unused/UnusedStubClass.class
[error] C:\Users\gnana\.ivy2\cache\org.apache.spark\spark-tags_2.11\jars\spark-tags_2.11-2.2.0.jar:org/apache/spark/unused/UnusedStubClass.class
[error] C:\Users\gnana\.ivy2\cache\org.spark-project.spark\unused\jars\unused-1.0.0.jar:org/apache/spark/unused/UnusedStubClass.class
[error]         at sbtassembly.Assembly$.applyStrategies(Assembly.scala:141)
[error]         at sbtassembly.Assembly$.x$1$lzycompute$1(Assembly.scala:25)
[error]         at sbtassembly.Assembly$.x$1$1(Assembly.scala:23)
[error]         at sbtassembly.Assembly$.stratMapping$lzycompute$1(Assembly.scala:23)
[error]         at sbtassembly.Assembly$.stratMapping$1(Assembly.scala:23)
[error]         at sbtassembly.Assembly$.inputs$lzycompute$1(Assembly.scala:67)
[error]         at sbtassembly.Assembly$.inputs$1(Assembly.scala:57)
[error]         at sbtassembly.Assembly$.apply(Assembly.scala:84)
[error]         at sbtassembly.Assembly$.$anonfun$assemblyTask$1(Assembly.scala:249)
[error]         at scala.Function1.$anonfun$compose$1(Function1.scala:44)
[error]         at sbt.internal.util.$tilde$greater.$anonfun$$u2219$1(TypeFunctions.scala:42)
[error]         at sbt.std.Transform$$anon$4.work(System.scala:64)
[error]         at sbt.Execute.$anonfun$submit$2(Execute.scala:257)
[error]         at sbt.internal.util.ErrorHandling$.wideConvert(ErrorHandling.scala:16)
[error]         at sbt.Execute.work(Execute.scala:266)
[error]         at sbt.Execute.$anonfun$submit$1(Execute.scala:257)
[error]         at sbt.ConcurrentRestrictions$$anon$4.$anonfun$submitValid$1(ConcurrentRestrictions.scala:167)
[error]         at sbt.CompletionService$$anon$2.call(CompletionService.scala:32)
[error]         at java.util.concurrent.FutureTask.run(FutureTask.java:266)
[error]         at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
[error]         at java.util.concurrent.FutureTask.run(FutureTask.java:266)
[error]         at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
[error]         at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
[error]         at java.lang.Thread.run(Thread.java:745)
[error] (*:assembly) deduplicate: different file contents found in the following:
[error] C:\Users\gnana\.ivy2\cache\org.apache.spark\spark-streaming-kafka-0-10_2.11\jars\spark-streaming-kafka-0-10_2.11-2.2.0.jar:org/apache/spark/unused/UnusedStubClass.class
[error] C:\Users\gnana\.ivy2\cache\org.apache.spark\spark-tags_2.11\jars\spark-tags_2.11-2.2.0.jar:org/apache/spark/unused/UnusedStubClass.class
[error] C:\Users\gnana\.ivy2\cache\org.spark-project.spark\unused\jars\unused-1.0.0.jar:org/apache/spark/unused/UnusedStubClass.class
[error] Total time: 91 s, completed Mar 11, 2018 6:15:45 PM

Answer 1

您需要在您的 SBT 文件中编写合并策略，这将帮助 SBT 为您选择正确的UnusedStubClass.class

organization := "com.example"
name := "cass-conn"
version := "0.1"
scalaVersion := "2.11.8"
val sparkVersion = "2.2.0"
val connectorVersion = "2.0.7"
val kafka_stream_version = "1.6.3"

libraryDependencies ++= Seq(
  "org.apache.spark" %% "spark-core" % sparkVersion  % "provided",
  "org.apache.spark" %% "spark-sql" % sparkVersion  % "provided",
  "org.apache.spark" %% "spark-hive" % sparkVersion  % "provided",
  "com.datastax.spark" %% "spark-cassandra-connector" % connectorVersion  ,
  "org.apache.spark" %% "spark-streaming-kafka-0-10" % "2.2.0",
  "org.apache.spark" %% "spark-streaming" %  "2.2.0"  % "provided",
)

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

Answer 2

检查您的 Java 版本，我在较高的 Java 版本上遇到了同样的问题，后来我降级到 Java8 来修复它