我有一个惊人的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集群:在线程“ main”中测试集群异常java.lang.NoClassDefFoundError:org / apache / spark / sql / Dataset at java.lang.Class.getDeclaredMethods0(本地方法),位于java.lang.Class.privateGetDeclaredMethods(未知源),位于java.io.ObjectStreamClass.getPrivateMethod(未知源),位于java.io.ObjectStreamClass.access $ 1700(未知) 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处的)。java.io.ObjectOutputStream.writeSerialData处的defaultWriteFields(未知源)java.io.ObjectOutputStream.writeOrdinaryObject处的Unknown Source(java.io.ObjectOutputStream.writeObject0处的未知源)(java.io.ObjectOutputStream.defaultWriteFields(未知源) java.io.ObjectOutputStream.writeSerialData上的Unknown Source(未知源)java.io.ObjectOutputStream.writeObject0上的java.io.ObjectOutputStream.writeOrdinaryObject(Unknown Source)上java.io.ObjectOutputStream.writeObject(未知源) ),位于scala.collection.immutable。$ colon $ colon.writeObject(List.scala:379),位于sun.reflect.NativeMethodAccessorImpl.invoke0(本地方法),位于sun.reflect.NativeMethodAccessorImpl.invoke(未知源),位于sun.reflect。 java.io上的DelegatingMethodAccessorImpl.invoke(未知源)(java.io处)。java.io.ObjectOutputStream.writeSerialData(未知源)(java.io.ObjectOutputStream.writeOrdinaryObject(未知源)(java.io.ObjectOutputStream.writeObject0)(未知源)(java.io.ObjectOutputStream)处的ObjectStreamClass.invokeWriteObject(未知源)。 java.io.ObjectOutputStream.writeSerialData处的defaultWriteFields(未知源)java.io.ObjectOutputStream.writeOrdinaryObject处的Unknown Source(java.io.ObjectOutputStream.writeObject0处的未知源)(java.io.ObjectOutputStream.defaultWriteFields(未知源) java.io.ObjectOutputStream.writeSerialData上的Unknown Source)(java.io.ObjectOutputStream.writeOrdinaryObject(Unknown Source)上的java.io.ObjectOutputStream.writeObject0(java.io.ObjectOutputStream.defaultWriteFields(Unknown Source)上的Unknown Source) ),位于java.io.ObjectOutputStream。org.apache.spark.serializer中的java.io.ObjectOutputStream.writeObject0(未知源)处的java.io.ObjectOutputStream.writeObject(未知源)处的writeSerialData(未知源)。 org.apache.spark.util.ClosureCleaner $ .ensureSerializable(ClosureCleaner.scala:295)上的org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100)上的JavaSerializationStream.writeObject(JavaSerializer.scala:43) 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:2037)在org.apache.spark.SparkContext.runJob(SparkContext.scala:1911)在org.apache.spark.rdd.RDD $$ anonfun $ collect $ 1.apply(RDD.scala:893)在org.apache.spark.rdd.RDDOperationScope $ .withScope(RDDOperationScope.scala:151)在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.AbstractJavaRDDLike.collect(JavaRDDLike.scala:45)位于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.net.URLClassLoader.findClass(java.lang.ClassNotFoundException:org.apache.spark.sql.Dataset(未知源),java.lang.ClassLoader sun.misc.Launcher $ AppClassLoader.loadClass(java.lang.ClassLoader.loadClass(未知源)的.loadClass(未知源)...58更多
我将pom.xml更新了,但是并没有解决错误。有人可以帮我解决这个问题吗?
谢谢!
更新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;
}
}
本文收集自互联网,转载请注明来源。
如有侵权,请联系 [email protected] 删除。
我来说两句