Dataframe не сохраняется в Cassandra

Question

У меня есть одно приложение с Spark (version 1.4.0) и Spark-Cassandra-connector (version 1.3.0-M1). В котором я пытаюсь сохранить один файл данных в таблице Cassandra, который имеет два столбца (итого, сообщение). И я уже создал таблицу в Cassandra с этими двумя столбцами.

Вот мой код,

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(); 

Но я получил это исключение,

[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) 

Answer 1

Соединитель версии 1.3.x не поддерживает Спарк 1.4.x мы работаем над 1.4.x релиз в этот момент, ожидайте этого очень скоро.