如何添加到 cassandra 中的现有值？答案

【问题标题】：How do i add to a existing value in cassandra?如何添加到 cassandra 中的现有值？
【发布时间】：2020-10-12 04:30:41
【问题描述】：

我正在编写一个流应用程序，它要求我将传入值添加或减去我的 cassandra 表中预先存在的值。我看过 cql 的批处理文档，但还没有找到解决我的问题的方法。这是更多现场的一个小例子：

预先存在的表

table:{
word:{'hello',
neg:{-0.5},
neu:{0.3},
pos:{0.2},
comp:{0.7}
}
}

传入值：

word:{'hello',
neu:{0.4}
}

在这里我需要添加 0.4 和 0.3 并重新插入表中。

【问题讨论】：

表是仅从 Spark 作业更新，还是也可以从外部更新？您在使用什么 - Spark Streaming 或 Spark Structured Streaming？您使用的是什么版本的 Spark Cassandra 连接器？
只能通过 spark 作业进行更新。我使用的是 spark-cassandra 连接器 2.5.0，是的，我使用的是 spark 流

标签： cassandra spark-streaming spark-cassandra-connector distributed-database

【解决方案1】：

您可以通过joining (on the primary key of the rows) your input data with data in Cassandra using the leftJoinWithCassandraTable实现加法，然后将输入数据与获取的数据进行加法，并将数据写回Cassandra。

类似这样的东西（取自我自己的代码）：

// this case class is matching to Cassandra table & input data...
case class Data(....)
val data = ...data_that_you_received_and_casted_to_Data_case_class...
val joined = data.leftJoinWithCassandraTable[Data]("ks", "table")
// perform update of existing values, and prepare new data
val summed = joined.map({ case (n: Data, c: Option[Data]) =>
 c match {
   // if there is no data in Cassandra, just return input data
   case None => Data(n)
   // there is data in Cassandra, do the sum
   case Some(s) =>
     Data(..., n.neu + s.neu)}
})
// and write updated/new values
summed.saveToCassandra("ks", "table")

附：我建议使用Spark Cassandra Connector 2.5.0 直接支持的 Spark Structured Streaming，以及所谓的数据帧直接连接。使用数据框时，您的代码会更简单且可能更优化。

【讨论】：