Hi again,
Not sure if this is the right thing to do, but I pulled down the latest
spark-cassandra-connector, built the jars, and added --jars flag with
spark-submit (and added to all workers).
It looks like it moves a little further, but now I have the following error
(with all file content the same):
14/10/28 04:43:44 INFO AppClient$ClientActor: Connecting to master
spark://ip-172-31-38-112:7077...
14/10/28 04:43:44 INFO SparkDeploySchedulerBackend: SchedulerBackend is
ready for scheduling beginning after reached minRegisteredResourcesRatio:
0.0
Exception in thread main java.lang.NoSuchMethodError:
com.datastax.spark.connector.streaming.DStreamFunctions.saveToCassandra(Ljava/lang/String;Ljava/lang/String;Lcom/datastax/spark/connector/SomeColumns;Lcom/datastax/spark/connector/writer/RowWriterFactory;)V
at
org.apache.spark.examples.streaming.CassandraWordCount$.main(CassandraWordCount.scala:52)
at
org.apache.spark.examples.streaming.CassandraWordCount.main(CassandraWordCount.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at org.apache.spark.deploy.SparkSubmit$.launch(SparkSubmit.scala:328)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:75)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
On Mon, Oct 27, 2014 at 9:22 PM, Harold Nguyen har...@nexgate.com wrote:
Hi Spark friends,
I'm trying to connect Spark Streaming into Cassandra by modifying the
NetworkWordCount.scala streaming example, and doing the make as few
changes as possible but having it insert data into Cassandra.
Could you let me know if you see any errors?
I'm using the spark-cassandra-connector, and I receive this error when I
submit my spark jar:
==
Exception in thread main java.lang.NoClassDefFoundError:
com/datastax/spark/connector/mapper/ColumnMapper
at
org.apache.spark.examples.streaming.CassandraWordCount.main(CassandraWordCount.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
at
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at org.apache.spark.deploy.SparkSubmit$.launch(SparkSubmit.scala:328)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:75)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Caused by: java.lang.ClassNotFoundException:
com.datastax.spark.connector.mapper.ColumnMapper
at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
at java.security.AccessController.doPrivileged(Native Method)
at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
... 8 more
===
And here is the class I'm using:
package org.apache.spark.examples.streaming
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.storage.StorageLevel
import com.datastax.spark.connector._
import com.datastax.spark.connector.streaming._
object CassandraWordCount {
def main(args: Array[String]) {
if (args.length 2) {
System.err.println(Usage: NetworkWordCount hostname port)
System.exit(1)
}
val conf = new SparkConf(true).set(spark.cassandra.connection.host,
ec2-54-191-235-127.us-west-2.compute.amazonaws.com
).setAppName(CassandraWordCount)
val sc = new SparkContext(spark://ip-172-31-38-112:7077, test,
conf)
StreamingExamples.setStreamingLogLevels()
// Create the context with a 1 second batch size
val ssc = new StreamingContext(sc, Seconds(1))
// Create a socket stream on target ip:port and count the
// words in input stream of \n delimited text (eg. generated by 'nc')
// Note that no duplication in storage level only for running locally.
// Replication necessary in distributed scenario for fault tolerance.
val lines = ssc.socketTextStream(args(0), args(1).toInt,
StorageLevel.MEMORY_AND_DISK_SER)
val words = lines.flatMap(_.split( ))
val wordCounts = words.map(x = (x, 1)).reduceByKey(_ + _)
//-- Writing it to Cassandra
wordCounts.saveToCassandra(test, kv, SomeColumns(key, value))
ssc.start()
ssc.awaitTermination()
}
}
===
Finally, here is my sbt build file:
==
name := Simple Streaming
version := 1.0
scalaVersion := 2.10.4
libraryDependencies ++= Seq(
org.apache.spark % spark-streaming_2.10 % 1.1.0,
com.datastax.spark %% spark-cassandra-connector % 1.1.0-alpha3
