更新cassandra行触发cassandra

Question

我正在使用spark 1.2和spark cassandra连接器1.2.3，我正在尝试更新表的一些行：

示例：

CREATE TABLE myTable ( 
a text, 
b text, 
c text, 
date timestamp, 
d text, 
e text static, 
f text static, 
PRIMARY KEY ((a, b, c), date, d) 
) WITH CLUSTERING ORDER BY (date ASC, d ASC)

val interactions = sc.cassandraTable[(String, String, String, DateTime, String, String)]("keySpace", "myTable"). 
select("a","b","c","date", "d", "e","f") 
val empty = interactions.filter(r => r._6 == null).cache() 
empty.count()

我只计算"e“包含null的行数，并用"b”的值替换它们。

 val update_inter = empty.map( r =>  (r._1,r._2, r._3, r._4, r._5, r._2)) 
 update_inter.saveToCassandra("keySpace", "myTable", SomeColumns("a","b","c","date", "d", "e", "f"))

当我签入cqlsh时，这是有效的，但是当我通过spark cassandra请求相同的行时，我仍然得到空值。

这是spark cassandra连接器中的bug吗？谢谢你的帮助。

Answer 1

当插入/更新发生时，Cassandra将插入或更新的数据的新的时间戳版本写入另一个SSTable中，而不是在适当的位置覆盖行。

您的Spark作业要么不更新现有行，而是写入新行，要么您的SSTables尚未将更改写入磁盘。如果要将结果写入新表，则null 'e‘列的计数将为零。

尝试使用nodetool flush命令并阅读以下内容：Cassandra Compaction

Answer 2

我猜.mode('append')是用来追加的。我正面临着类似的问题，但使用的是java连接器，但在python中似乎可以使用此选项。