Spark和Cassandra Java应用程序：线程“ main”中的异常java.lang.NoClassDefFoundError：org / apache / spark / sql / Dataset

Question

我有一个惊人的siplme
Java应用程序，几乎可以从这个示例中复制它：http
://markmail.org/download.xqy?id=zua6upabiylzeetp&number=2

我要做的就是读取表数据并在Eclipse控制台中显示。

我的pom.xml：

        <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>chat_connaction_test</groupId>
  <artifactId>ChatSparkConnectionTest</artifactId>
  <version>0.0.1-SNAPSHOT</version>
 <dependencies> 
    <dependency>
    <groupId>com.datastax.cassandra</groupId>
    <artifactId>cassandra-driver-core</artifactId>
    <version>3.1.0</version>
    </dependency>

    <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.10</artifactId>
    <version>2.0.0</version>
    </dependency>

    <!-- https://mvnrepository.com/artifact/com.datastax.spark/spark-cassandra-connector_2.10 -->
    <dependency>
    <groupId>com.datastax.spark</groupId>
    <artifactId>spark-cassandra-connector_2.10</artifactId>
    <version>2.0.0-M3</version>
    </dependency>

    <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming_2.10 -->
    <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming_2.10</artifactId>
    <version>2.0.0</version>
    </dependency>
    <!--
    <dependency> 
    <groupId>org.apache.spark</groupId> 
    <artifactId>spark-hive_2.10</artifactId> 
    <version>1.5.2</version> 
    </dependency>
    -->
  </dependencies>
</project>

而我的Java代码：

    package com.chatSparkConnactionTest;

import static com.datastax.spark.connector.japi.CassandraJavaUtil.javaFunctions;
import java.io.Serializable;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import com.datastax.spark.connector.japi.CassandraRow;

public class JavaDemo implements Serializable {
    private static final long serialVersionUID = 1L;
    public static void main(String[] args) {

        SparkConf conf = new SparkConf().
            setAppName("chat").
            setMaster("local").
            set("spark.executor.memory","1g").
            set("spark.cassandra.connection.host", "127.0.0.1");
        JavaSparkContext sc = new JavaSparkContext(conf);

        JavaRDD<String> cassandraRowsRDD = javaFunctions(sc).cassandraTable(
            "chat", "dictionary")

            .map(new Function<CassandraRow, String>() {
                @Override
                public String call(CassandraRow cassandraRow) throws Exception {
                    String tempResult = cassandraRow.toString();
                    System.out.println(tempResult);
                    return tempResult;
                    }
                }
            );
        System.out.println("Data as CassandraRows: \n" + 
        cassandraRowsRDD.collect().size()); // THIS IS A LINE WITH ERROR
    } 
}

这是我的错误：

16/10/05
20:49:18信息CassandraConnector：连接到Cassandra群集：在线程“主”中测试群集异常java.lang.NoClassDefFoundError：org
/ apache / spark / sql / Dataset at java.lang.Class.getDeclaredMethods0（
java.io.ObjectStreamClass.access $
1700处的java.lang.Class.privateGetDeclaredMethods（未知源）的java.io.ObjectStreamClass.getPrivateMethod（未知源）的java.lang.Class.privateGetDeclaredMethods（未知源）
Source）在java.io.ObjectStreamClass $ 2.run（未知源）在java.io.ObjectStreamClass $
2.run（在Java.security.AccessController.doPrivileged（本机方法）在java.io.ObjectStreamClass。
）的java.io.ObjectStreamClass.lookup（Unknown
Source）处的java.io.ObjectOutputStream.writeObject0（Unknown
Source）处的java）。io.ObjectOutputStream.defaultWriteFields（未知源）（java.io.ObjectOutputStream.writeSerialData（未知源），java.io.ObjectOutputStream.writeOrdinaryObject（未知源），java.io.ObjectOutputStream.writeObject0（未知源），java.io。
java.io.ObjectOutputStream.writeSerialData（未知源）（java.io.ObjectOutputStream.writeOrdinaryObject（未知源），java.io.ObjectOutputStream.writeObject0（未知源），java.io.ObjectOutputStream。位置scala.collection.immutable。$
colon $
colon.writeObject（List.scala：379）处的writeObject（未知源）位置sun.reflect.NativeMethodAccessorImpl.invoke0（本机方法）处sun.reflect.NativeMethodAccessorImpl.invoke（未知源）处sun.reflect.DelegatingMethodAccessorImpl。java.io.ObjectOutputStream.writeSerialData（未知源）处的java.io.ObjectStreamClass.invokeWriteObject（未知源）处的java.lang.reflect.Method.invoke（未知源）处的invoke（未知源）。
java.io.ObjectOutputStream.writeObject0（未知源）上的writeOrdinaryObject（未知源）java.io.ObjectOutputStream.writeSerialData处java.io.ObjectOutputStream.writeSerialData（未知源）处java.io.ObjectOutputStream.writeOrdinaryObject（未知源）
java.io.ObjectOutputStream.writeObject0上的未知源（java.io.ObjectOutputStream.defaultWriteFields（未知源）上的java.io.ObjectOutputStream.writeSerialData（未知源）上的java.io.ObjectOutputStream.writeOrdinaryObject（未知源）
），位于java.io。java.io.ObjectOutputStream.defaultWriteFields（未知源）处的ObjectOutputStream.writeObject0（未知源）java.io.ObjectOutputStream.writeOrdinaryObject（未知源处）的java.io.ObjectOutputStream.writeSerialData（未知源）处。
org.apache.spark.serializer.JavaSerializerInstance.serialize（JavaSerializer.java.io.ObjectOutputStream.writeObject（Unknown
Source）的org.apache.spark.serializer.JavaSerializationStream.writeObject（JavaSerializer.scala：43）处的writeObject0（Unknown
Source） .scala：100），位于org.apache.spark.util.ClosureCleaner $
.ensureSerializable（ClosureCleaner.scala：295），位于org.apache.spark.util.ClosureCleaner
$ .org $ apache $ spark $ util $ ClosureCleaner $$ clean（
ClosureCleaner.scala：288），位于org.apache.spark.util.ClosureCleaner $
.clean（ClosureCleaner.scala：108）位于org.apache.spark.SparkContext.runJob（SparkContext.scala：1896）位于org.apache.spark.SparkContext.runJob（SparkContext.scala：1896）位于org.apache.spark.SparkContext.runJob（SparkContext.scala
：1911）at org.apache.spark.rdd.RDD $$ anonfun $ collect $
1.apply（RDD.scala：893）at org.apache.spark.rdd.RDDOperationScope $
.withScope（RDDOperationScope.scala：151）at org
org.apache.spark.rdd.RDD.withScope（RDD.scala：358）上的.apache.spark.rdd.RDDOperationScope
$ .withScope（RDDOperationScope.scala：112）在org.apache.spark.rdd.RDD.collect（
RDD.scala：892）位于org.apache.spark.api.java.JavaRDDLike $
class.collect（JavaRDDLike.scala：360）位于org.apache.spark.api.java.AbstractJavaRDDLike.collect（JavaRDDLike.scala：45）
com.chatSparkConnactionTest.JavaDemo.main（JavaDemo.java:37）上的原因：java.lang.ClassNotFoundException：org。java.net.URLClassLoader.findClass上的apache.spark.sql.Dataset（java.lang.ClassLoader.loadClass（未知源）上的sun.misc.Launcher
$ AppClassLoader.loadClass（java.lang。 ClassLoader.loadClass（未知来源）… 58和

我将我的pom.xml更新了，但是并没有解决错误。有人可以帮我解决这个问题吗？

谢谢！

更新1：这是我的构建路径屏幕截图： 链接到我的屏幕截图

Answer 1

您收到“ java.lang.NoClassDefFoundError：org / apache / spark / sql /
Dataset”错误，因为pom.xml文件中缺少“ spark-sql”依赖项。

如果要使用Spark 2.0.0读取Cassandra表，则需要以下最低依赖项。

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.11</artifactId>
    <version>2.0.0</version>
</dependency>
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.11</artifactId>
    <version>2.0.0</version>
</dependency>
<dependency>
    <groupId>com.datastax.spark</groupId>
    <artifactId>spark-cassandra-connector_2.11</artifactId>
    <version>2.0.0-M3</version>
</dependency>

Spark 2.0.0提供了SparkSession和数据集API。下面是读取Cassandra表并打印记录的示例程序。

 public class SparkCassandraDatasetApplication {
 public static void main(String[] args) {
 SparkSession spark = SparkSession
          .builder()
          .appName("SparkCassandraDatasetApplication")
          .config("spark.sql.warehouse.dir", "/file:C:/temp")
          .config("spark.cassandra.connection.host", "127.0.0.1")
          .config("spark.cassandra.connection.port", "9042")
          .master("local[2]")
          .getOrCreate();

 //Read data
 Dataset<Row> dataset = spark.read().format("org.apache.spark.sql.cassandra")
        .options(new HashMap<String, String>() {
            {
                put("keyspace", "mykeyspace");
                put("table", "mytable");
            }
        }).load();

   //Print data
   dataset.show();       
   spark.stop();
   }        
}

如果仍然要使用RDD，请使用下面的示例程序。

public class SparkCassandraRDDApplication {
public static void main(String[] args) {
    SparkConf conf = new SparkConf()
            .setAppName("SparkCassandraRDDApplication")
            .setMaster("local[2]")
            .set("spark.cassandra.connection.host", "127.0.0.1")
            .set("spark.cassandra.connection.port", "9042");

    JavaSparkContext sc = new JavaSparkContext(conf);

    //Read
    JavaRDD<UserData> resultsRDD = javaFunctions(sc).cassandraTable("mykeyspace", "mytable",CassandraJavaUtil.mapRowTo(UserData.class));

    //Print
    resultsRDD.foreach(data -> {
        System.out.println(data.id);
        System.out.println(data.username);
    });

    sc.stop();
  }
}

上面程序中使用的Javabean（UserData）如下。

public class UserData implements Serializable{  
  String id;
  String username;     
  public String getId() {
      return id;
  }
  public void setId(String id) {
      this.id = id;
  }
  public String getUsername() {
     return username;
  }
  public void setUsername(String username) {
     this.username = username;
   }    
}