Spark java.lang.NoClassDefFoundError 中的 spark-cassandra-connector 出错：com/datastax/driver/core/ProtocolOptions$Compression

Question

当我尝试使用 spark-cassandra-connector 连接到 cassandra 时出现此错误：

线程“main”中的异常 java.lang.NoClassDefFoundError: com/datastax/driver/core/ProtocolOptions$Compression 在 com.datastax.spark.connector.cql.CassandraConnectorConf$.(CassandraConnectorConf.scala:112) 在 com.datastax.spark.connector.cql.CassandraConnectorConf$.(CassandraConnectorConf.scala) 在 com.datastax.spark.connector.cql.CassandraConnector$.apply(CassandraConnector.scala:192) 在 com.datastax.spark.connector.SparkContextFunctions.cassandraTable$default$3(SparkContextFunctions.scala:48) 在 main.scala.TestSpark$.main(TestSpark.scala:19) 在 main.scala.TestSpark.main(TestSpark.scala) 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:606) 在 org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:672) 在 org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:180) 在 org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:205) 在 org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:120) 在 org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala) 引起：java.lang.ClassNotFoundException：com.datastax.driver.core.ProtocolOptions$Compression 在 java.net.URLClassLoader$1.run(URLClassLoader.java:366) 在 java.net.URLClassLoader$1.run(URLClassLoader.java:355) 在 java.security.AccessController.doPrivileged（本机方法） 在 java.net.URLClassLoader.findClass(URLClassLoader.java:354) 在 java.lang.ClassLoader.loadClass(ClassLoader.java:425) 在 sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308) 在 java.lang.ClassLoader.loadClass(ClassLoader.java:358) ... 15 更多 我在 spark 类路径中添加了 jar spark-cassandra-connector_2.11-1.5.0-M2.jar

我已经在sbt文件中添加了依赖：

name := "Simple Project"

version := "1.0"

scalaVersion := "2.11.7"

libraryDependencies += "org.apache.spark" %% "spark-core" % "1.5.1"

libraryDependencies += "org.apache.spark" %% "spark-sql" % "1.5.1"

libraryDependencies += "com.datastax.spark" %% "spark-cassandra-connector" % "1.5.0-M2"

libraryDependencies += "com.datastax.spark" %% "spark-cassandra-connector-java" % "1.5.0-M2"

这是我尝试执行的 scala 程序：

package main.scala


import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import com.datastax.spark.connector._

/**
 * Created by Simo on 01.12.15.
 */
object TestSpark {
  def main(args: Array[String]) {
   val conf = new SparkConf(true)
        .set("spark.cassandra.connection.host", "54.229.218.236")
        .setAppName("Simple Application")
    val sc= new SparkContext("local", "test", conf)
    val rdd = sc.cassandraTable("test", "kv")
    println(rdd.count)
    println(rdd.first)
    println(rdd.map(_.getInt("value")).sum)
  }
}

这就是我运行它的方式：

$ sbt package
$ $SPARK_HOME/bin/spark-submit --class "main.scala.TestSpark" target/scala-2.11/simple-project_2.11-1.0.jar

你能帮我理解我做错了什么吗？

谢谢！

编辑：

我已尝试在依赖项列表和 spark 类路径中添加 Datastax 驱动程序：

libraryDependencies += "com.datastax.cassandra" % "cassandra-driver-core" % "2.1.9"
libraryDependencies += "com.datastax.cassandra" % "cassandra-driver-mapping" % "2.1.9"

最后一个错误不再出现，但现在我又有一个错误：

线程“main”中的异常 java.lang.NoSuchMethodError: scala.runtime.ObjectRef.zero()Lscala/runtime/ObjectRef; 在 com.datastax.spark.connector.cql.CassandraConnector$.com$datastax$spark$connector$cql$CassandraConnector$$createSession(CassandraConnector.scala) 在 com.datastax.spark.connector.cql.CassandraConnector$$anonfun$2.apply(CassandraConnector.scala:150) 在 com.datastax.spark.connector.cql.CassandraConnector$$anonfun$2.apply(CassandraConnector.scala:150) 在 com.datastax.spark.connector.cql.RefCountedCache.createNewValueAndKeys(RefCountedCache.scala:31) 在 com.datastax.spark.connector.cql.RefCountedCache.acquire(RefCountedCache.scala:56) 在 com.datastax.spark.connector.cql.CassandraConnector.openSession(CassandraConnector.scala:81) 在 com.datastax.spark.connector.cql.CassandraConnector.withSessionDo(CassandraConnector.scala:109) 在 com.datastax.spark.connector.cql.CassandraConnector.withClusterDo(CassandraConnector.scala:120) 在 com.datastax.spark.connector.cql.Schema$.fromCassandra(Schema.scala:241) 在 com.datastax.spark.connector.rdd.CassandraTableRowReaderProvider$class.tableDef(CassandraTableRowReaderProvider.scala:51) 在 com.datastax.spark.connector.rdd.CassandraTableScanRDD.tableDef$lzycompute(CassandraTableScanRDD.scala:59) 在 com.datastax.spark.connector.rdd.CassandraTableScanRDD.tableDef(CassandraTableScanRDD.scala:59) 在 com.datastax.spark.connector.rdd.CassandraTableRowReaderProvider$class.verify(CassandraTableRowReaderProvider.scala:146) 在 com.datastax.spark.connector.rdd.CassandraTableScanRDD.verify(CassandraTableScanRDD.scala:59) 在 com.datastax.spark.connector.rdd.CassandraTableScanRDD.getPartitions(CassandraTableScanRDD.scala:143) 在 org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239) 在 org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237) 在 scala.Option.getOrElse(Option.scala:120) 在 org.apache.spark.rdd.RDD.partitions(RDD.scala:237) 在 org.apache.spark.SparkContext.runJob(SparkContext.scala:1919) 在 org.apache.spark.rdd.RDD.count(RDD.scala:1121) 在 main.scala.TestSpark$.main(TestSpark.scala:20) 在 main.scala.TestSpark.main(TestSpark.scala) 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:606) 在 org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:672) 在 org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:180) 在 org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:205) 在 org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:120) 在 org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

编辑 2：在编译时制作 scala 2.10.6（与 spark 的 scala 版本相同） 以前的错误不再出现，但我有这个新错误：

线程“主”java.lang.NoClassDefFoundError 中的异常：com/google/common/util/concurrent/AsyncFunction 在 com.datastax.spark.connector.cql.DefaultConnectionFactory$.clusterBuilder(CassandraConnectionFactory.scala:36) 在 com.datastax.spark.connector.cql.DefaultConnectionFactory$.createCluster(CassandraConnectionFactory.scala:85) 在 com.datastax.spark.connector.cql.CassandraConnector$.com$datastax$spark$connector$cql$CassandraConnector$$createSession(CassandraConnector.scala:155) 在 com.datastax.spark.connector.cql.CassandraConnector$$anonfun$2.apply(CassandraConnector.scala:150) 在 com.datastax.spark.connector.cql.CassandraConnector$$anonfun$2.apply(CassandraConnector.scala:150) 在 com.datastax.spark.connector.cql.RefCountedCache.createNewValueAndKeys(RefCountedCache.scala:31) 在 com.datastax.spark.connector.cql.RefCountedCache.acquire(RefCountedCache.scala:56) 在 com.datastax.spark.connector.cql.CassandraConnector.openSession(CassandraConnector.scala:81) 在 com.datastax.spark.connector.cql.CassandraConnector.withSessionDo(CassandraConnector.scala:109) 在 com.datastax.spark.connector.cql.CassandraConnector.withClusterDo(CassandraConnector.scala:120) 在 com.datastax.spark.connector.cql.Schema$.fromCassandra(Schema.scala:241) 在 com.datastax.spark.connector.rdd.CassandraTableRowReaderProvider$class.tableDef(CassandraTableRowReaderProvider.scala:51) 在 com.datastax.spark.connector.rdd.CassandraTableScanRDD.tableDef$lzycompute(CassandraTableScanRDD.scala:59) 在 com.datastax.spark.connector.rdd.CassandraTableScanRDD.tableDef(CassandraTableScanRDD.scala:59) 在 com.datastax.spark.connector.rdd.CassandraTableRowReaderProvider$class.verify(CassandraTableRowReaderProvider.scala:150) 在 com.datastax.spark.connector.rdd.CassandraTableScanRDD.verify(CassandraTableScanRDD.scala:59) 在 com.datastax.spark.connector.rdd.CassandraTableScanRDD.getPartitions(CassandraTableScanRDD.scala:143) 在 org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:239) 在 org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:237) 在 scala.Option.getOrElse(Option.scala:120) 在 org.apache.spark.rdd.RDD.partitions(RDD.scala:237) 在 org.apache.spark.SparkContext.runJob(SparkContext.scala:1919) 在 org.apache.spark.rdd.RDD.count(RDD.scala:1121) 在 main.scala.TestSpark$.main(TestSpark.scala:20) 在 main.scala.TestSpark.main(TestSpark.scala) 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:606) 在 org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:672) 在 org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:180) 在 org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:205) 在 org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:120) 在 org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala) 引起：java.lang.ClassNotFoundException：com.google.common.util.concurrent.AsyncFunction 在 java.net.URLClassLoader$1.run(URLClassLoader.java:366) 在 java.net.URLClassLoader$1.run(URLClassLoader.java:355) 在 java.security.AccessController.doPrivileged（本机方法） 在 java.net.URLClassLoader.findClass(URLClassLoader.java:354) 在 java.lang.ClassLoader.loadClass(ClassLoader.java:425) 在 sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308) 在 java.lang.ClassLoader.loadClass(ClassLoader.java:358) ... 34 更多

Answer 1

最终按照@Odomontois 的建议使用 sbt-assembly 解决

这是最终的 build.sbt：

name := "Simple Project"

version := "1.0"

scalaVersion := "2.10.6"

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

libraryDependencies += "org.apache.spark" %% "spark-sql" % "1.5.1" % "provided"

libraryDependencies += "com.datastax.cassandra" % "cassandra-driver-core" % "2.1.9"

libraryDependencies += "com.datastax.spark" % "spark-cassandra-connector_2.10" % "1.5.0-M2"



jarName in assembly :="my-project-assembly.jar"

assemblyOption in assembly := (assemblyOption in             assembly).value.copy(includeScala = false)


resolvers += "Akka Repository" at "http://repo.akka.io/releases/"

mergeStrategy in assembly <<= (mergeStrategy in assembly) { (old) =>
    {
        case PathList("netty", "handler", xs @ _*)         => MergeStrategy.first
        case PathList("netty", "buffer", xs @ _*)     => MergeStrategy.first
        case PathList("netty", "common", xs @ _*)     => MergeStrategy.first
        case PathList("netty", "transport", xs @ _*)     => MergeStrategy.first
        case PathList("netty", "codec", xs @ _*)     => MergeStrategy.first

        case PathList("META-INF", "io.netty.versions.properties") => MergeStrategy.first
        case x => old(x)
        }
    }

Answer 2

您还需要从（根据 Spark-cassandra 连接器的版本）添加 Datastax Cassandra 驱动程序的依赖项：- https://repo1.maven.org/maven2/com/datastax/cassandra/cassandra-driver-core/