将Spark streaming PySpark dataframe写入Cassandra会覆盖表而不是追加

Question

我正在运行一个由Kafka、Spark和Cassandra组成的1节点集群。全部本地在同一台机器上。

从一个简单的Python脚本中，我每5秒将一些虚拟数据流到一个Kafka主题中。然后使用Spark结构化流，我将这个数据流（一次一行）读入PySpark DataFrame中，并使用startingoffset=lates。最后，我尝试将此行追加到一个已经存在的Cassandra表中。

我一直在关注（如何向Cassandra编写流数据集？）和（来自Kafka主题的PySpark结构流的Cassandra Sink）。

有一行数据被成功写入Cassandra表，但我的问题是它每次都被覆盖，而不是追加到表的末尾。我可能做错了什么？

下面是我的代码：

CQL DDL用于在Cassandra中创建KafKasparkkeyspace，后跟RandintStream表：

DESCRIBE keyspaces;

CREATE KEYSPACE kafkaspark
  WITH REPLICATION = { 
   'class' : 'SimpleStrategy', 
   'replication_factor' : 1 
  };
  
USE kafkaspark; 

CREATE TABLE randIntStream (
    key int,
    value int,
    topic text,
    partition int,
    offset bigint,
    timestamp timestamp,
    timestampType int,
    PRIMARY KEY (partition, topic)
);

./bin/pyspark --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.0.1,com.datastax.spark:spark-cassandra-connector_2.12:3.0.0 --conf spark.cassandra.connection.host=127.0.0.1,spark.sql.extensions=com.datastax.spark.connector.CassandraSparkExtensions

df = spark.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:9092").option("startingOffsets","latest").option("subscribe","topic1").load()

df2 = df.withColumn("key", df["key"].cast("string")).withColumn("value", df["value"].cast("string"))
df3 = df2.withColumn("key", df2["key"].cast("integer")).withColumn("value", df2["value"].cast("integer"))
df4 = df3.withColumnRenamed("timestampType","timestamptype")
df4.printSchema()

def writeToCassandra(writeDF, epochId):
    writeDF.write \
    .format("org.apache.spark.sql.cassandra") \
    .options(table="randintstream", keyspace="kafkaspark") \
    .mode("append") \
    .save()

最后，从Spark向Cassandra写查询：

query = df4.writeStream \
.trigger(processingTime="5 seconds") \
.outputMode("update") \
.foreachBatch(writeToCassandra) \
.start()

Answer 1

如果行总是在Cassandra中重写，那么表中的主键可能不正确--您需要确保每一行都有唯一的主键。如果您是从Spark创建Cassandra表，那么默认情况下它只是将第一列作为分区键，而且它可能不是唯一的。

提供架构后更新：

是的，这就是我提到的情况-主键为(partition,topic)，但是从该主题读取的特定分区中的每一行都具有相同的主键值，因此它将覆盖以前的版本。您需要使主键唯一-例如，将offset或timestamp列添加到主键（尽管如果在同一毫秒内生成数据，timestamp可能不唯一）。

df4.writeStream \
  .trigger(processingTime="5 seconds") \
  .format("org.apache.spark.sql.cassandra") \
  .options(table="randintstream", keyspace="kafkaspark") \
  .mode("update") \
  .start()