如何使用Datastax连接器从Spark Dataframe更新特定的Cassandra列集

Question

我有一个只有几个列的Cassandra表，我想更新其中的一个（对于多个列，还有什么？）来自Spark 2.4.0。但是如果我没有提供所有的列，那么记录就不会得到更新。

Cassandra模式：

rowkey,message,number,timestamp,name
1,hello,12345,12233454,ABC

finalDF.select("rowkey","current_ts")
  .withColumnRenamed("current_ts","timestamp")
  .write
  .format("org.apache.spark.sql.cassandra")
  .options(Map("table" -> "table_data", "keyspace" -> "ks_data"))
  .mode("overwrite")
  .option("confirm.truncate","true")
  .save()

finalDF=
rowkey,current_ts
1,12233999

rowkey,message,number,timestamp,name
1,hello,12345,12233999,ABC

Answer 1

说明是savemode用于指定将DataFrame保存到数据源的预期行为。（不仅适用于C*，而且适用于任何数据源）。可用的选项有

1. savemode.errorifexists
2. savemode.append
3. savemode.overwrite
4. 保存。忽略

在本例中，由于您已经有了数据，并且希望追加数据，因此必须使用savemode.append

import org.apache.spark.sql.SaveMode

finalDF.select("rowkey","current_ts")
  .withColumnRenamed("current_ts","timestamp")
  .write
  .format("org.apache.spark.sql.cassandra")
  .options(Map("table" -> "table_data", "keyspace" -> "ks_data"))
  .mode(SaveMode.Append)
  .option("confirm.truncate","true")
  .save()

请参阅SaveModes上的spark文档