使用java中的Apache Spark连接器从Cassandra中检索数据的错误

Question

我有使用sparkCassandraConnector从Cassandra表中检索数据的问题。我在卡桑德拉创建了名为“ks”的名称空间和表“学生”。该表如下：

id |名称

---- + -----------

10 |凯瑟琳

我开始在本地运行星火start-all.sh

然后我创造了这个类“SparkCassandraConnector”，它具有用于连接火花和Cassandra.What我试图做一个命令是从获取数据学生表并将其打印在屏幕上。

我得到的错误是“抛出java.lang.ClassNotFoundException：SparkCassandraConnector $学生 java.net.URLClassLoader的$ 1.run（URLClassLoader.java:372） java.net.URLClassLoader的$ 1.run（URLClassLoader.java:361 ） java.security.AccessController.doPrivileged（本机方法） java.net.URLClassLoader.findClass（URLClassLoader.java:360） java.lang.ClassLoader.loadClass（ClassLoader.java:424） java.lang.ClassLoader中。 loadClass（ClassLoader.java:357） java.lang.Class.forName0（Native Method） java.lang.Class.forName（Class.java:340）

这是我的计划：

import org.apache.commons.lang.StringUtils; 

import org.apache.spark.SparkConf; 

import org.apache.spark.api.java.JavaRDD; 

import org.apache.spark.api.java.JavaSparkContext; 

import java.io.Serializable; 

import static com.datastax.spark.connector.CassandraJavaUtil.javaFunctions; 

public class SparkCassandraConnector implements Serializable { 
public static void main(String[] args) { 

    SparkConf conf = new SparkConf().setAppName("Simple Application"); 

    conf.setMaster("spark://127.0.0.1:7077"); 
    conf.set("spark.cassandra.connection.host", "127.0.0.1"); 
    String[] jars = new String[10]; 
    jars[0] = "~/.m2/repository/com/datastax/spark/spark-cassandra-connector-java_2.10/1.1.0-alpha4/spark-cassandra-connector-java_2.10-1.1.0-alpha4.jar"; 
    jars[1] = "~/.m2/repository/com/datastax/cassandra/cassandra-driver-core/2.1.0/cassandra-driver-core-2.1.0.jar"; 
    jars[3] = "~/.m2/repository/com/datastax/spark/spark-cassandra-connector_2.10/1.1.0-alpha4/spark-cassandra-connector_2.10-1.1.0-alpha4.jar"; 
    jars[4] = "~/.m2/repository/com/datastax/cassandra/cassandra-driver-core/2.1.0/cassandra-driver-core-2.1.0.jar"; 
    jars[5] = "~/.m2/repository/org/apache/cassandra/cassandra-thrift/2.1.0/cassandra-thrift-2.1.0.jar"; 
    jars[6] = "~/.m2/repository/org/apache/cassandra/cassandra-clientutil/2.1.0/cassandra-clientutil-2.1.0.jar"; 
    conf = conf.setJars(jars); 
    JavaSparkContext sc = new JavaSparkContext(conf); 

    JavaRDD<String> rdd = javaFunctions(sc).cassandraTable("ks", "student", Student.class) 
      .map(new org.apache.spark.api.java.function.Function<Student, String>() { 
       @Override 
       public String call(Student person) throws Exception { 
        return person.toString(); 
       } 
      }); 
    System.out.println("Data as Person beans: \n" + StringUtils.join(rdd.collect(), "\n")); 
} 
public static class Student implements Serializable{ 

    private Integer id; 
    private String name; 

    public Student(){ 

    } 
    public Student(Integer id, String name) { 
     this.id = id; 
     this.name = name; 
    } 

    public Integer getId() { 
     return id; 
    } 

    public void setId(Integer id) { 
     this.id = id; 
    } 

    public String getName() { 
     return name; 
    } 

    public void setName(String name) { 
     this.name = name; 
    } 
} 

}

这是我的POM文件：

<dependencies> 


    <!--Spark--> 

    <dependency> 
     <groupId>org.apache.spark</groupId> 
     <artifactId>spark-core_2.10</artifactId> 
     <version>1.1.0</version> 
    </dependency> 
    <dependency> 
     <groupId>org.apache.spark</groupId> 
     <artifactId>spark-streaming_2.10</artifactId> 
     <version>1.1.0</version> 
    </dependency> 
    <dependency> 
     <groupId>com.datastax.cassandra</groupId> 
     <artifactId>cassandra-driver-core</artifactId> 
     <version>2.1.0</version> 
    </dependency> 
    <dependency> 
     <groupId>com.datastax.spark</groupId> 
     <artifactId>spark-cassandra-connector_2.10</artifactId> 
     <version>1.1.0-alpha4</version> 
    </dependency> 

    <dependency> 
     <groupId>com.datastax.spark</groupId> 
     <artifactId>spark-cassandra-connector-java_2.10</artifactId> 
     <version>1.1.0-alpha4</version> 
    </dependency> 

    <dependency> 
     <groupId>org.apache.spark</groupId> 
     <artifactId>spark-catalyst_2.10</artifactId> 
     <version>1.0.0</version> 
    </dependency> 
</dependencies> 

Answer 1

在提供的罐子，含有活的罐子，因此Student.class丢失。快速修复它以添加项目文件夹中的jar。

另一种方法是将工作和所有依赖关系打包在“超级jar”中，并将该超级jar用作唯一声明的jar。看看maven shade plugin.

也可以从命令行使用spark-submit --jars选项提供罐。