apache-spark - java.lang.UnsupportedOperationException : 'Writing to a non-empty Cassandra Table is not allowed

Question

我有一个场景，我将接收由我的 Spark 流程序处理的流数据，并且每个间隔的输出被附加到我现有的 cassandra 表中。

目前，我的 Spark 流程序将生成一个数据框，我需要将其保存在我的 cassandra 表中。我目前面临的问题是当我使用下面的命令时，我无法将数据/行附加到我现有的 cassandra 表中

dff.write.format("org.apache.spark.sql.cassandra").options(Map("table" -> "xxx", "yyy" -> "retail")).save()

我已阅读以下链接 http://rustyrazorblade.com/2015/08/migrating-from-mysql-to-cassandra-using-spark/他将 mode="append"传递到 save 方法中，但它抛出语法错误

我也无法从下面的链接中了解我需要在哪里修复
https://groups.google.com/a/lists.datastax.com/forum/#!topic/spark-connector-user/rlGGWQF2wnM

需要有关如何解决此问题的帮助。我正在 Scala 中编写我的 Spark 流作业

Answer 1

我认为你必须通过以下方式做到这一点:

dff.write.format("org.apache.spark.sql.cassandra").mode(SaveMode.Append).options(Map("table" -> "xxx", "yyy" -> "retail")).save()

cassandra 处理数据的方式迫使您执行所谓的“更新插入”——您必须记住，插入可能会覆盖某些已存储记录的主键与插入记录的主键相同的行。 Cassandra 是一个“快速写入”数据库，因此它在写入之前不会检查数据是否存在。