I have one application with Spark (version 1.4.0) and Spark-Cassandra-connector (version 1.3.0-M1). In which, I am trying to store one dataframe into Cassandra table which has two columns (total, message). And i already created table into Cassandra with these two columns.
Here is my Code,
scoredTweet.foreachRDD(new Function2<JavaRDD<Message>,Time,Void>(){
@Override
public Void call(JavaRDD<Message> arg0, Time arg1) throws Exception {
SQLContext sqlContext = SparkConnection.getSqlContext();
DataFrame df = sqlContext.createDataFrame(arg0, Message.class);
df.registerTempTable("messages");
DataFrame aggregatedMessages = sqlContext.sql("select count(*) as total,message from messages group by message");
aggregatedMessages.show();
aggregatedMessages.printSchema();
aggregatedMessages.write().mode(SaveMode.Append)
.option("keyspace", Properties.getString("spark.cassandra.keyspace"))
.option("c_table", Properties.getString("spark.cassandra.aggrtable"))
.format("org.apache.spark.sql.cassandra").save();
But I got this exception,
[Stage 20:===========================> (103 + 2) / 199]
[Stage 20:====================================> (134 + 2) / 199]
[Stage 20:============================================> (164 + 2) / 199]
[Stage 20:====================================================> (193 + 2) / 199]
+-----+--------------------+
|total| message|
+-----+--------------------+
| 1|there is deep pol...|
| 1|RT @SwarupPhD: Ag...|
| 1|#3Novices : #Desp...|
| 1|RT @Babu_Bhaiyaa:...|
| 1|https://t.co/BMPX...|
+-----+--------------------+
root
|-- total: long (nullable = false)
|-- message: string (nullable = true)
15/06/12 21:24:40 INFO Cluster: New Cassandra host /192.168.1.17:9042 added
15/06/12 21:24:40 INFO Cluster: New Cassandra host /192.168.1.19:9042 added
15/06/12 21:24:40 INFO LocalNodeFirstLoadBalancingPolicy: Added host 192.168.1.19 (datacenter1)
15/06/12 21:24:40 INFO Cluster: New Cassandra host /192.168.1.21:9042 added
15/06/12 21:24:40 INFO LocalNodeFirstLoadBalancingPolicy: Added host 192.168.1.21 (datacenter1)
15/06/12 21:24:40 INFO CassandraConnector: Connected to Cassandra cluster: BDI Cassandra
15/06/12 21:24:41 INFO CassandraConnector: Disconnected from Cassandra cluster: BDI Cassandra
15/06/12 21:26:14 ERROR JobScheduler: Error running job streaming job 1434124380000 ms.1
java.util.NoSuchElementException: key not found: frozen<tuple<int, text, text, text, list<text>>>
at scala.collection.MapLike$class.default(MapLike.scala:228)
at scala.collection.AbstractMap.default(Map.scala:58)
at scala.collection.MapLike$class.apply(MapLike.scala:141)
at scala.collection.AbstractMap.apply(Map.scala:58)
at com.datastax.spark.connector.types.ColumnType$.fromDriverType(ColumnType.scala:73)
at com.datastax.spark.connector.types.ColumnType$$anonfun$1.apply(ColumnType.scala:67)
at com.datastax.spark.connector.types.ColumnType$$anonfun$1.apply(ColumnType.scala:67)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
at scala.collection.Iterator$class.foreach(Iterator.scala:727)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
at scala.collection.AbstractTraversable.map(Traversable.scala:105)
at com.datastax.spark.connector.types.ColumnType$.fromDriverType(ColumnType.scala:67)
at com.datastax.spark.connector.cql.ColumnDef$.apply(Schema.scala:110)
at com.datastax.spark.connector.cql.Schema$$anonfun$com$datastax$spark$connector$cql$Schema$$fetchRegularColumns$1.apply(Schema.scala:210)
at com.datastax.spark.connector.cql.Schema$$anonfun$com$datastax$spark$connector$cql$Schema$$fetchRegularColumns$1.apply(Schema.scala:206)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
at scala.collection.AbstractTraversable.map(Traversable.scala:105)
at com.datastax.spark.connector.cql.Schema$.com$datastax$spark$connector$cql$Schema$$fetchRegularColumns(Schema.scala:206)
at com.datastax.spark.connector.cql.Schema$$anonfun$com$datastax$spark$connector$cql$Schema$$fetchTables$1$2.apply(Schema.scala:235)
at com.datastax.spark.connector.cql.Schema$$anonfun$com$datastax$spark$connector$cql$Schema$$fetchTables$1$2.apply(Schema.scala:232)
at scala.collection.TraversableLike$WithFilter$$anonfun$map$2.apply(TraversableLike.scala:722)
at scala.collection.immutable.Set$Set2.foreach(Set.scala:94)
at scala.collection.TraversableLike$WithFilter.map(TraversableLike.scala:721)
at com.datastax.spark.connector.cql.Schema$.com$datastax$spark$connector$cql$Schema$$fetchTables$1(Schema.scala:232)
at com.datastax.spark.connector.cql.Schema$$anonfun$com$datastax$spark$connector$cql$Schema$$fetchKeyspaces$1$2.apply(Schema.scala:241)
at com.datastax.spark.connector.cql.Schema$$anonfun$com$datastax$spark$connector$cql$Schema$$fetchKeyspaces$1$2.apply(Schema.scala:240)
at scala.collection.TraversableLike$WithFilter$$anonfun$map$2.apply(TraversableLike.scala:722)
at scala.collection.immutable.HashSet$HashSet1.foreach(HashSet.scala:153)
at scala.collection.immutable.HashSet$HashTrieSet.foreach(HashSet.scala:306)
at scala.collection.immutable.HashSet$HashTrieSet.foreach(HashSet.scala:306)
at scala.collection.TraversableLike$WithFilter.map(TraversableLike.scala:721)
at com.datastax.spark.connector.cql.Schema$.com$datastax$spark$connector$cql$Schema$$fetchKeyspaces$1(Schema.scala:240)
at com.datastax.spark.connector.cql.Schema$$anonfun$fromCassandra$1.apply(Schema.scala:246)
at com.datastax.spark.connector.cql.Schema$$anonfun$fromCassandra$1.apply(Schema.scala:243)
at com.datastax.spark.connector.cql.CassandraConnector$$anonfun$withClusterDo$1.apply(CassandraConnector.scala:116)
at com.datastax.spark.connector.cql.CassandraConnector$$anonfun$withClusterDo$1.apply(CassandraConnector.scala:115)
at com.datastax.spark.connector.cql.CassandraConnector$$anonfun$withSessionDo$1.apply(CassandraConnector.scala:105)
at com.datastax.spark.connector.cql.CassandraConnector$$anonfun$withSessionDo$1.apply(CassandraConnector.scala:104)
at com.datastax.spark.connector.cql.CassandraConnector.closeResourceAfterUse(CassandraConnector.scala:156)
at com.datastax.spark.connector.cql.CassandraConnector.withSessionDo(CassandraConnector.scala:104)
at com.datastax.spark.connector.cql.CassandraConnector.withClusterDo(CassandraConnector.scala:115)
at com.datastax.spark.connector.cql.Schema$.fromCassandra(Schema.scala:243)
at org.apache.spark.sql.cassandra.CassandraSourceRelation.<init>(CassandraSourceRelation.scala:39)
at org.apache.spark.sql.cassandra.CassandraSourceRelation$.apply(CassandraSourceRelation.scala:168)
at org.apache.spark.sql.cassandra.DefaultSource.createRelation(DefaultSource.scala:84)
at org.apache.spark.sql.sources.ResolvedDataSource$.apply(ddl.scala:305)
at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:144)
The connector version 1.3.x does not support Spark 1.4.x we are working on a 1.4.x release at this moment, expect it very soon.