在Cassandra中用新的键值对更新地图类型列，而不是完全覆盖地图。

Posted 2023-02-17

继续提出以下问题 将Spark数据集[(String，Map[String，String])]插入Cassandra表。.

我有一个Spark数据集，类型是Dataset[(String，Map[String，String])]。

我必须将其插入到一个Cassandra表中。

这里，Dataset[(String，Map[String，String])]中的key将成为我在Cassandra中的行的主键。

Dataset[(String，Map[String，String])]中的Map将进入同一行的ColumnNameValueMap列中。

我的Cassandra表结构是。

CREATE TABLE SampleKeyspace.CassandraTable (
  RowKey text PRIMARY KEY,
  ColumnNameValueMap map<text,text>
);

我能够使用Spark Cassandra连接器在Cassandra表中插入数据。

现在，我正在更新同一个map列（第2列），为同一个rowkey（第1列primary key）添加新的键值。但是，对这一列的每次新更新都会清除之前的地图。

如何使用Spark Cassandra连接器追加同一个地图？

答案

我不认为可以直接从Dataframe API中做，但我认为可以从Dataframe API中做。可以通过RDD API来实现。. 例如，我有以下tabble与一些测试数据。

CREATE TABLE test.m1 (
    id int PRIMARY KEY,
    m map<int, text>
);
cqlsh> select * from test.m1;                                                                                                                                                                                                  id | m                                                                                                        ----+--------------------                                                                                        1 | 1: 't1', 2: 't2'                                                                                                                                                                                                      (1 rows)

我有数据在Spark中。

scala> val data = Seq((1, Map(3 -> "t3"))).toDF("id", "m")                                                     data: org.apache.spark.sql.DataFrame = [id: int, m: map<int,string>]

然后，我可以指定我想把数据附加到特定的列，用下面的代码。

data.rdd.saveToCassandra("test", "m1", SomeColumns("id", "m" append))

然后我就可以看到数据被更新了。

cqlsh> select * from test.m1;
id | m
----+----------------------------- 
 1 | 1: 't1', 2: 't2', 3: 't3'
(1 rows)

此外 append的元素，支持删除带有 remove 选项，以及 prepend (仅适用于列表)。 文件中载有这方面的例子。