数据框未保存到 Cassandra 中

Question

我有一份申请 Spark (version 1.4.0) 和 Spark-Cassandra-connector (version 1.3.0-M1)。其中，我试图将一个数据帧存储到具有两列（总计，消息）的 Cassandra table 中。我已经用这两列在 Cassandra 中创建了 table。

这是我的代码，

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.apply(ColumnType.scala:67)
    at com.datastax.spark.connector.types.ColumnType$$anonfun.apply(ColumnType.scala:67)
    at scala.collection.TraversableLike$$anonfun$map.apply(TraversableLike.scala:244)
    at scala.collection.TraversableLike$$anonfun$map.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.apply(Schema.scala:210)
    at com.datastax.spark.connector.cql.Schema$$anonfun$com$datastax$spark$connector$cql$Schema$$fetchRegularColumns.apply(Schema.scala:206)
    at scala.collection.TraversableLike$$anonfun$map.apply(TraversableLike.scala:244)
    at scala.collection.TraversableLike$$anonfun$map.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.apply(Schema.scala:235)
    at com.datastax.spark.connector.cql.Schema$$anonfun$com$datastax$spark$connector$cql$Schema$$fetchTables.apply(Schema.scala:232)
    at scala.collection.TraversableLike$WithFilter$$anonfun$map.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(Schema.scala:232)
    at com.datastax.spark.connector.cql.Schema$$anonfun$com$datastax$spark$connector$cql$Schema$$fetchKeyspaces.apply(Schema.scala:241)
    at com.datastax.spark.connector.cql.Schema$$anonfun$com$datastax$spark$connector$cql$Schema$$fetchKeyspaces.apply(Schema.scala:240)
    at scala.collection.TraversableLike$WithFilter$$anonfun$map.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(Schema.scala:240)
    at com.datastax.spark.connector.cql.Schema$$anonfun$fromCassandra.apply(Schema.scala:246)
    at com.datastax.spark.connector.cql.Schema$$anonfun$fromCassandra.apply(Schema.scala:243)
    at com.datastax.spark.connector.cql.CassandraConnector$$anonfun$withClusterDo.apply(CassandraConnector.scala:116)
    at com.datastax.spark.connector.cql.CassandraConnector$$anonfun$withClusterDo.apply(CassandraConnector.scala:115)
    at com.datastax.spark.connector.cql.CassandraConnector$$anonfun$withSessionDo.apply(CassandraConnector.scala:105)
    at com.datastax.spark.connector.cql.CassandraConnector$$anonfun$withSessionDo.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 不支持 Spark 1。4.x我们正在开发 1。4.x 目前正在发布，期待很快。

数据框未保存到 Cassandra 中

Dataframe is not saved into Cassandra

java

cassandra

apache-spark

apache-spark-sql

spark-cassandra-connector