作业提交时的 Scala fat jar 依赖问题

2023-12-12

我使用 Scala 编写了简单的 kafka 流。它在本地运行良好。我已经采取了 fat jar 并在 scala 集群中提交。提交作业后，我收到“未找到类”错误。如果我提取 fat jar，它在 fat jar 内具有所有依赖项。

为什么我收到类未找到错误？如何解决这个问题？

Note:如果我手动将 fat jar 部署（复制）到 Spark/jars 文件夹中。我没有看到任何问题。但是，这不是正确的做法

我正在使用 Windows 7 并在同一台计算机上运行主节点和工作节点。

工作提交

spark-2.2\bin>spark-submit --class Welcome --master spark://169.254.208.125:7077 C:\Gnana\cass-conn-assembly-0.1.jar

Code

import org.apache.spark.streaming.{Milliseconds, Seconds, StreamingContext}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import com.datastax.spark.connector._, org.apache.spark.SparkContext, org.apache.spark.SparkConf
import com.datastax.spark.connector.streaming._


object Welcome {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("demo").setMaster("spark://169.254.208.125:7077");
    conf.set("spark.cassandra.connection.host", "192.168.1.2")
    val sc = new SparkContext(conf)

    sc.setLogLevel("WARN")
    val ssc = new StreamingContext(sc, Milliseconds(100))
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "192.168.1.2:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "use_a_separate_group_id_for_each_stream",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )
    val topics = Array("test")
    val messages = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )
    val lines = messages.map(_.value)
    lines.map(li=>{val arr= li.split(",");(arr(0).toInt,arr(1),arr(2),arr(3))}).saveToCassandra("inventory", "emp",SomeColumns("emp_id","create_date","emp_city","emp_name"))
    println(" Spark is ready !!!!!! ");

    /*sys.ShutdownHookThread {
      println("Gracefully stopping Spark Streaming Application")
      ssc.stop(stopSparkContext = true, stopGracefully = true)
      println("Application stopped")
    }*/

    ssc.start();
    ssc.awaitTermination();
  }
  def sayHello(msg:String): Unit = {
    print("welcome to Sacala "+msg);
  }
}

构建.sbt

organization := "com.demo"
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)
}

Error 1:

Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/kafka/clients/consumer/Consumer
        at org.apache.spark.streaming.kafka010.ConsumerStrategies$.Subscribe(ConsumerStrategy.scala:256)
        at Welcome$.main(Welcome.scala:32)
        at Welcome.main(Welcome.scala)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:498)
        at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.sca
        at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:180)
        at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:205)
        at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:119)
        at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Caused by: java.lang.ClassNotFoundException: org.apache.kafka.clients.consumer.Consumer
        at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
        at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
        at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
        ... 12 more

构建日志

[info] Done updating.
[info] Compiling 2 Scala sources to C:\Gnana\cass-conn\target\scala-2.11\classes ...
[info] Done compiling.
[info] Including: slf4j-api-1.7.21.jar
[info] Including: joda-time-2.3.jar
[info] Including: kafka_2.11-0.10.1.0.jar
[info] Including: scala-library-2.11.8.jar
[info] Including: jopt-simple-4.9.jar
[info] Including: metrics-core-2.2.0.jar
[info] Including: slf4j-log4j12-1.7.21.jar
[info] Including: log4j-1.2.17.jar
[info] Including: joda-convert-1.2.jar
[info] Including: zkclient-0.9.jar
[info] Including: scala-reflect-2.11.8.jar
[info] Including: zookeeper-3.4.8.jar
[info] Including: jline-0.9.94.jar
[info] Including: netty-3.7.0.Final.jar
[info] Including: lz4-1.3.0.jar
[info] Including: scala-parser-combinators_2.11-1.0.4.jar
[info] Including: snappy-java-1.1.2.6.jar
[info] Including: kafka-clients-0.10.1.0.jar
[info] Including: spark-streaming-kafka-0-10_2.11-2.2.0.jar
[info] Including: spark-tags_2.11-2.2.0.jar
[info] Including: unused-1.0.0.jar
[info] Including: netty-all-4.0.33.Final.jar
[info] Including: commons-beanutils-1.9.3.jar
[info] Including: jsr166e-1.1.0.jar
[info] Including: spark-cassandra-connector_2.11-2.0.7.jar
[info] Including: commons-collections-3.2.2.jar
[info] Checking every *.class/*.jar file's SHA-1.
[info] Merging files...
[warn] Merging 'NOTICE' with strategy 'rename'
[warn] Merging 'META-INF\NOTICE.txt' with strategy 'rename'
[warn] Merging 'META-INF\NOTICE' with strategy 'rename'
[warn] Merging 'org\xerial\snappy\native\README' with strategy 'rename'
[warn] Merging 'META-INF\LICENSE.txt' with strategy 'rename'
[warn] Merging 'META-INF\license' with strategy 'rename'
[warn] Merging 'LICENSE.txt' with strategy 'rename'
[warn] Merging 'META-INF\LICENSE' with strategy 'rename'
[warn] Merging 'LICENSE' with strategy 'rename'
[warn] Merging 'META-INF\DEPENDENCIES' with strategy 'discard'
[warn] Merging 'META-INF\INDEX.LIST' with strategy 'discard'
[warn] Merging 'META-INF\MANIFEST.MF' with strategy 'discard'
[warn] Merging 'META-INF\maven\com.datastax.cassandra\cassandra-driver-core\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\com.datastax.cassandra\cassandra-driver-core\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\com.datastax.cassandra\cassandra-driver-mapping\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\com.datastax.cassandra\cassandra-driver-mapping\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\com.github.jnr\jffi\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\com.github.jnr\jffi\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\com.github.jnr\jnr-constants\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\com.github.jnr\jnr-constants\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\com.github.jnr\jnr-ffi\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\com.github.jnr\jnr-ffi\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\com.github.jnr\jnr-posix\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\com.github.jnr\jnr-posix\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\com.github.jnr\jnr-x86asm\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\com.github.jnr\jnr-x86asm\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\com.google.guava\guava\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\com.google.guava\guava\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\com.twitter\jsr166e\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\com.twitter\jsr166e\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\com.yammer.metrics\metrics-core\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\com.yammer.metrics\metrics-core\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\commons-beanutils\commons-beanutils\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\commons-beanutils\commons-beanutils\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\commons-collections\commons-collections\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\commons-collections\commons-collections\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\io.netty\netty-all\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\io.netty\netty-all\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\io.netty\netty\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\io.netty\netty\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\jline\jline\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\jline\jline\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\joda-time\joda-time\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\joda-time\joda-time\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\log4j\log4j\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\log4j\log4j\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\net.sf.jopt-simple\jopt-simple\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\net.sf.jopt-simple\jopt-simple\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\org.apache.spark\spark-streaming-kafka-0-10_2.11\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\org.apache.spark\spark-streaming-kafka-0-10_2.11\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\org.apache.spark\spark-tags_2.11\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\org.apache.spark\spark-tags_2.11\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\org.joda\joda-convert\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\org.joda\joda-convert\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\org.slf4j\slf4j-api\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\org.slf4j\slf4j-api\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\org.slf4j\slf4j-log4j12\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\org.slf4j\slf4j-log4j12\pom.xml' with strategy 'discard'
[warn] Merging 'META-INF\maven\org.spark-project.spark\unused\pom.properties' with strategy 'discard'
[warn] Merging 'META-INF\maven\org.spark-project.spark\unused\pom.xml' with strategy 'discard'
[warn] Merging 'org\apache\spark\unused\UnusedStubClass.class' with strategy 'first'
[warn] Strategy 'discard' was applied to 51 files
[warn] Strategy 'first' was applied to a file
[warn] Strategy 'rename' was applied to 9 files
[info] SHA-1: 85f8513511b46290883ab70f2525b04a8d3c33d7

您缺少 kafka 依赖项：

"org.apache.kafka" %% "kafka" % "0.10.1.0"

添加这一点，你就会一切都好。

我有我的build.sbt像这样设置：

"org.apache.kafka" %% "kafka"                      % "0.10.1.0",
"org.apache.spark" %% "spark-streaming-kafka-0-10" % "2.2.0",
"org.apache.spark" %% "spark-streaming"            % "2.2.0" % "provided"

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

scala

sbt

SparkStreaming

作业提交时的 Scala fat jar 依赖问题的相关文章

IntelliJ IDEA 13 给出有关不兼容类型的无效错误（Play 中的 Java-Scala-InterOp）

我刚刚从 IDEA 12 升级到 13 社区版从那时起我在 IDEA 中收到关于我的游戏项目的类型不兼容的错误 Option
为什么 astyanax (java) 无法识别我的 scala 案例类参数列表中的 @Id 注解值？

所以这是我的困境我有一个域模型其中有一堆 scala 中的案例类例如User and Organization 在我的数据访问层 dao 存储库等中我使用 astyanax 来自 netflix 的 java 库及其实体持久器将
具有动态命名参数的 Scala 案例类副本

对于具有参数数量的 scala 案例类 21 e g case class Car type String brand String door Int 其中类型吉普车品牌丰田门 4 etc 还有一个复制方法允许使用命名参数覆盖 Ca
如何通过 Scala REPL 使用第三方库？

我已经下载了Algebird我想尝试一些事情Scala 解释器使用这个库我该如何实现这一目标当然您可以使用 scala cp 无论如何并手动管理您的依赖项但这变得非常乏味尤其是当您有多个依赖项时更灵活的方法是使用sbt http
将额外的参数传递给多态函数？

我有一个多态函数可以将列表转换为集合 import shapeless PolyDefns gt import shapeless val lists List 1 2 List A B List 1 1 2 2 HNil object
从会话中读取数据时如何发出加特林捕获请求？

根据加特林文档 http gatling io docs 2 1 7 session session api html 我可以在执行场景时使用会话属性但是每次我在场景中使用函数文字访问会话时都会遇到以下异常 error java la
在 scala 中保留推导的更高类型

我有一个高阶类型并致力于用它构建一些 DSL 我正在寻找一种方法来定义可以接受类型而无需显式指定此类型的函数自我描述示例 class Wrap T val data T class DSL def doSomething T x Wra
Scala 'null' 是否算作另一种类型的实例？

我有这个代码 class MyLinkedList T h T tail MyLinkedList T def prepend v T MyLinkedList T new MyLinkedList v this 我想知道我如何可以将第二个
用 HashMap[Int, Vector[Int]] (Scala) 表示图（邻接列表）？

我想知道如何如果可能的话我可以通过以下方式制作可变图的邻接列表表示HashMap Int Vector Int HashMap当然是可变的目前我将其设置为HashMap Int ArrayBuffer Int 但我可以更改 Arr
如何在 akka actor 中测试公共方法？

我有一个 akka 演员 class MyActor extends Actor def recieve def getCount id String Int do a lot of stuff proccess id do more st
Spark Streaming 中是否需要检查点

我注意到 Spark 流示例也有检查点代码我的问题是检查点有多重要如果是为了容错那么在此类流应用程序中发生故障的频率是多少这一切都取决于您的用例假设您正在运行一个流作业它仅从 Kafka 读取数据并计算记录数如果您的应用程序在
Scala：将整个列表的 Either 与每个元素的 Either 组合

我有一个 Either 列表它代表错误 type ErrorType List String type FailFast A Either ErrorType A import cats syntax either val l List
理解 scala 的 _ 与 Any/Nothing

如果一个类具有协变类型参数例如Iterable A http www scala lang org archives downloads distrib files nightly docs 2 10 1 library index ht
错误：无法在 scala 中找到或加载主类

安装 eclipse scala 插件和 eclipse maven scala 插件后我是 scala 新手所以我尝试确保在测试 scala hello world 项目后环境正常工作它按预期工作但我在尝试执行我从公司存储库中签出
如何使用 FS2 中的分类器函数对对象进行分组？

我有一个无序的流measurements 我想将其分组为固定大小的批次以便以后可以有效地保留它们 val measurements for id lt Seq foo bar baz value lt 1 to 5 yield id va
SBT 对 Scala 类型感到困惑

SBT 抛出以下错误 value split is not a member of String String error filter arg gt arg split delimiter length gt 2 对于以下代码块 impl
对 Spark 数据集中的数字字符串进行排序

假设我有以下内容Dataset productCode amount XX 13 300 XX 1 250 XX 2 410 XX 9 50 XX 10 35 XX 100 870 Where productCode is of Strin
相当于 scala 中的 python repr()

有没有相当于Python的东西reprscala 中的函数即您可以给任何 scala 对象提供一个函数它将生成该对象的字符串表示形式该对象是有效的 scala 代码 eg val l List Map 1 gt a print re
Jack（Java Android 编译器套件）将如何影响 Scala 开发人员

现在随着公告Jack https source android com source jack html谷歌阐明了 Java 与 Android 相关的可预见的未来但这对 Scala 和其他基于 JVM 的语言开发人员有何影响尤其 Sc
承诺的反面是什么？

承诺代表将来可能可用或无法实现的值我正在寻找的是一种数据类型它表示将来可能变得不可用的可用值可能是由于错误 Promise a b TransitionFromTo

随机推荐

SAS 宏，将值作为字符串传递给 where 子句

我下面有一个 SAS 宏不起作用此代码片段不返回任何值因为 where 语句不起作用有人有主意吗我尝试添加 str 但这也不起作用 macro refreshments beverage type proc sql select w
WPF DataGrid 性能 - 筛选性能

我们正处于将应用程序从 SilverLight 转换为 WPF 的阶段我们在其中开发了我们自己的自定义 DataGrid 位于 Silverlight 的本机 DataGrid 之上应用程序由大量视图组成其中近 99 的视图都在使
如何在SQL中根据多个“标签”查询数据？

我有三个简单的表 Items ItemID int PK ItemName nvarchar50 ItemCost int Tags TagID int PK TagName nvarchar50 ItemTags ItemID int F
Foreach 循环（或者 do while 可能？） - 只想根据页面返回一条记录

使用下面编辑的代码解决了这个问题感谢所有提供帮助的人我的数据库中有两条记录每条记录有 6 个字段 challengeId partnerName code challengeTitle description image url 我从
解析 CSV 文件并聚合值、多列

我想修改这里的帖子解析 CSV 文件并聚合值对多列而不是一列求和所以对于这些数据 CITY AMOUNT AMOUNT2 AMOUNTn London 20 21 22 Tokyo 45 46 47 London 55 56 57 N
使用reduce 或join 组合数组有什么区别？

考虑以下字符串数组 let arrayStrings H e l l o 为了组合它的元素将 Hello 作为单个字符串我们可以 reduce it let reducedString arrayStrings reduce 0 1 H
如何在 Ajax 调用中添加 UI 旋转器？

我想在 Ajax 调用上显示加载微调器我试过spin js图书馆但它不起作用这是我的 JavaScript 函数它使用 Ajax 调用 function sendRequest ajax url spinner type get c
如果剧集直接连续或重叠，则合并日期行

我有一个这样的表 ID BEGIN END 如果同一 ID 存在重叠剧集例如2000 01 01 2001 12 31 and 2000 06 01 2002 06 31 我想使用合并行MIN BEGIN MAX END 如果剧集是直接连
单击键盘中的“完成”按钮时如何调用方法？

我想在 UITextField KeyBoard 中单击完成按钮时调用一个方法请帮我甚至没有必要实现委托我非常喜欢使用良好的老式目标操作模式来处理这个问题如果您有多种结束编辑的方法例如拦截文本字段外部的触摸以取消编辑它还
根据 Avalonia 中的 DataContext 属性选择 DataTemplate

我正在实现一个 UserControl 它应该显示设置列表 public class SettingPropertyItem string Name get Type ValueType get object Value get set 基
我无法安装 Netbeans10

作为一个java初学者我发现安装Netbeans8很容易但安装Netbeans10却很困难似乎我需要处理我下载的文档这是Netbeans网站上的说明但我不明白为什么我不能安装它请帮我知道如何处理它要从源代码构建 Apache
LinkedIn API 中的目标受众计数

LinkedIn API v1 有一个 API 端点来计算所选目标受众的数量后面的api只有这个端点Ads API 现在我尝试使用定位标准来分享帖子分享API 我想限制发帖的受众人数少于300人由于LinkedIn有限制受众人数必
检测 Xamarin Scrollview 何时到达末尾

我正在使用 xamarin 表单在滚动视图中显示项目列表我想添加一个功能以便当用户滚动到页面底部时我可以触发一个事件例如加载更多项目目前我正在使用此 OnScrolled 函数来触发加载事件 private void OnScro
为多个函数正确实现（并发）asyncio

我有几个函数它们的执行并不相互依赖我想做的是同时执行它们而不是顺序执行同步我也添加了事件循环但我无法确定它是否正常工作这是实现 File 1 import file2 def funcA a b 1 2 file2 main a
当我们运行可执行文件时，所有部分都会立即加载到内存中吗？

因此可执行文件包含不同的部分和标头 At the ELF Header我们可以看到一些关于它们的元数据例如不同标头的大小起始点等可执行文件的不同部分是否会同时加载到内存中 If yes 怎样几时它被定义了我们可以在哪里看到相关信
Java中如何读写XML文件并在保存时将注释节点视为文本节点

我正在读取从外部系统检索的 Java XML 文件然后对其进行处理最终将其保存在本地并将其部署回来外部系统给我一个包含此节点的 XML 文件
C : 创建一个仅使用“write”函数来显示数字的函数

我怎样才能编写一个基本函数只需包含在 write 函数中 include
如何使用 swift 5 URLSession 中引入的新结果类型？

Swift 5 引入了新的 Result 类型来处理异步函数的结果我想知道如何在 URLSession 中使用这个新的结果类型我有以下代码 func getCategorByAPI Base Url is from an static
出现错误“不允许更改属性‘osDisk.name’。”使用 Azure Powershell 脚本

我正在尝试将带有托管磁盘的现有 Azure VM 移至现有可用性集中但是当我应用命令时 New AzureRmVM ResourceGroupName rg Location OriginalVM Location VM NewVM D
作业提交时的 Scala fat jar 依赖问题

我使用 Scala 编写了简单的 kafka 流它在本地运行良好我已经采取了 fat jar 并在 scala 集群中提交提交作业后我收到未找到类错误如果我提取 fat jar 它在 fat jar 内具有所有依赖项为什么我

作业提交时的 Scala fat jar 依赖问题

作业提交时的 Scala fat jar 依赖问题 的相关文章

随机推荐

热门标签

作业提交时的 Scala fat jar 依赖问题的相关文章