更新 Cassandra 行时的奇怪行为

Question

我正在使用pyspark 和pyspark-cassandra。

我注意到在多个版本的 Cassandra（3.0.x 和 3.6.x）上使用 COPY、sstableloader 和现在在 saveToCassandra 中使用 pyspark 的这种行为。

我有以下架构

CREATE TABLE test (
    id int,
    time timestamp,
    a int,
    b int,
    c int,
    PRIMARY KEY ((id), time)
) WITH CLUSTERING ORDER BY (time DESC);

以及以下数据

(1, datetime.datetime(2015, 3, 1, 0, 18, 18, tzinfo=<UTC>), 1, 0, 0)
(1, datetime.datetime(2015, 3, 1, 0, 19, 12, tzinfo=<UTC>), 0, 1, 0)
(1, datetime.datetime(2015, 3, 1, 0, 22, 59, tzinfo=<UTC>), 1, 0, 0)
(1, datetime.datetime(2015, 3, 1, 0, 23, 52, tzinfo=<UTC>), 0, 1, 0)
(1, datetime.datetime(2015, 3, 1, 0, 32,  2, tzinfo=<UTC>), 1, 1, 0)
(1, datetime.datetime(2015, 3, 1, 0, 32,  8, tzinfo=<UTC>), 0, 2, 0)
(1, datetime.datetime(2015, 3, 1, 0, 43, 30, tzinfo=<UTC>), 1, 1, 0)
(1, datetime.datetime(2015, 3, 1, 0, 44, 12, tzinfo=<UTC>), 0, 2, 0)
(1, datetime.datetime(2015, 3, 1, 0, 48, 49, tzinfo=<UTC>), 1, 1, 0)
(1, datetime.datetime(2015, 3, 1, 0, 49,  7, tzinfo=<UTC>), 0, 2, 0)
(1, datetime.datetime(2015, 3, 1, 0, 50,  5, tzinfo=<UTC>), 1, 1, 0)
(1, datetime.datetime(2015, 3, 1, 0, 50, 53, tzinfo=<UTC>), 0, 2, 0)
(1, datetime.datetime(2015, 3, 1, 0, 51, 53, tzinfo=<UTC>), 1, 1, 0)
(1, datetime.datetime(2015, 3, 1, 0, 51, 59, tzinfo=<UTC>), 0, 2, 0)
(1, datetime.datetime(2015, 3, 1, 0, 54, 35, tzinfo=<UTC>), 1, 1, 0)
(1, datetime.datetime(2015, 3, 1, 0, 55, 28, tzinfo=<UTC>), 0, 2, 0)
(1, datetime.datetime(2015, 3, 1, 0, 55, 55, tzinfo=<UTC>), 1, 2, 0)
(1, datetime.datetime(2015, 3, 1, 0, 56, 24, tzinfo=<UTC>), 0, 3, 0)
(1, datetime.datetime(2015, 3, 1, 1, 11, 14, tzinfo=<UTC>), 1, 2, 0)
(1, datetime.datetime(2015, 3, 1, 1, 11, 17, tzinfo=<UTC>), 2, 1, 0)
(1, datetime.datetime(2015, 3, 1, 1, 12,  8, tzinfo=<UTC>), 1, 2, 0)
(1, datetime.datetime(2015, 3, 1, 1, 12, 10, tzinfo=<UTC>), 0, 3, 0)
(1, datetime.datetime(2015, 3, 1, 1, 17, 43, tzinfo=<UTC>), 1, 2, 0)
(1, datetime.datetime(2015, 3, 1, 1, 17, 49, tzinfo=<UTC>), 0, 3, 0)
(1, datetime.datetime(2015, 3, 1, 1, 24, 12, tzinfo=<UTC>), 1, 2, 0)
(1, datetime.datetime(2015, 3, 1, 1, 24, 18, tzinfo=<UTC>), 2, 1, 0)
(1, datetime.datetime(2015, 3, 1, 1, 24, 18, tzinfo=<UTC>), 1, 2, 0)
(1, datetime.datetime(2015, 3, 1, 1, 24, 24, tzinfo=<UTC>), 2, 1, 0)

在数据的末尾，有两行具有相同的时间戳。

(1, datetime.datetime(2015, 3, 1, 1, 24, 18, tzinfo=<UTC>), 2, 1, 0)
(1, datetime.datetime(2015, 3, 1, 1, 24, 18, tzinfo=<UTC>), 1, 2, 0)

据我了解，当我保存到 Cassandra 时，其中一个会“获胜” - 只有一排。

使用 cassandra 写入后

rdd.saveToCassandra(keyspace, table, ['id', 'time', 'a', 'b', 'c'])

似乎两行都没有获胜。相反，这些行似乎已经“合并”了。

   1 |  2015-03-01 01:17:43+0000 |        1 |        2 |        0
   1 |  2015-03-01 01:17:49+0000 |        0 |        3 |        0
   1 |  2015-03-01 01:24:12+0000 |        1 |        2 |        0
   1 |  2015-03-01 01:24:18+0000 |        2 |        2 |        0
   1 |  2015-03-01 01:24:24+0000 |        2 |        1 |        0

而不是2015-03-01 01:24:18+0000包含(1, 2, 0)或(2, 1, 0)，它包含(2, 2, 0)。

这里发生了什么？我一生都无法弄清楚这种行为是由什么引起的。

Answer 1

这是一个鲜为人知的效果，它来自于数据的批处理。批处理写入为批处理中的所有插入分配相同的时间戳。接下来，如果使用 exact 相同的时间戳完成两次写入，则存在一个特殊的合并规则，因为没有“最后一次”写入。 Spark Cassandra 连接器默认使用分区内批处理，因此如果您有这种类型的值破坏，很可能会发生这种情况。

两个相同写入时间戳的行为是基于较大值的合并。

给定表（键，a，b）

Batch
Insert "foo", 2, 1
Insert "foo", 1, 2
End batch

批次为两个突变提供相同的时间戳。 Cassandra 不能选择“最后写入”，因为它们同时发生，而是选择两者中较大的值。合并后的结果是

"foo", 2, 2