我想把spark结构化流数据写入cassandra。我的spark版本是2.4.0。
我研究了一些post和datasax企业平台的一些用法。我没用过，找到了方法 foreachBatch 这有助于将流数据写入接收器。
我审查了一个基于databricks网站的文档。自己试试。
这是我写的代码：

parsed = parsed_opc \
    .withWatermark("sourceTimeStamp", "10 minutes") \
    .dropDuplicates(["id", "sourceTimeStamp"]) \
    .groupBy(
        window(parsed_opc.sourceTimeStamp, "4 seconds"),
        parsed_opc.id
    ) \
    .agg({"value": "avg"}) \
    .withColumnRenamed("avg(value)", "avg")\
    .withColumnRenamed("window", "sourceTime") 

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

parsed.writeStream \
    .foreachBatch(writeToCassandra) \
    .outputMode("update") \
    .start()

的架构 parsed Dataframe是：

root
 |-- sourceTime: struct (nullable = false)
 |    |-- start: timestamp (nullable = true)
 |    |-- end: timestamp (nullable = true)
 |-- id: string (nullable = true)
 |-- avg: double (nullable = true)

我可以像这样成功地将这个流式df写入控制台：

query = parsed \
  .writeStream \
  .format("console")\
  .outputMode("complete")\
  .start()

控制台输出如下：

+--------------------+----+---+
|          sourceTime|  id|avg|
+--------------------+----+---+
|[2019-07-20 18:55...|Temp|2.0|
+--------------------+----+---+

所以，当写入控制台时，这没关系。但当我在 cqlsh 表中没有附加记录。
这是cassandra中的table create脚本：

CREATE TABLE poc.opc ( id text, avg float,sourceTime timestamp PRIMARY KEY );

那么，你能告诉我怎么了吗？