Как обновить определенный набор столбцов Cassandra из Spark Dataframe, используя соединитель Datastax - PullRequest
Новая
       50

Как обновить определенный набор столбцов Cassandra из Spark Dataframe, используя соединитель Datastax

0 голосов
/ 26 декабря 2018

У меня есть таблица Cassandra из нескольких столбцов, и я хочу обновить один из них (а также что для нескольких столбцов?) Из Spark 2.4.0.Но если я не предоставлю все столбцы, то записи не будут обновляться.

Схема Кассандры: 

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

Точка Spark DataFrame состоит из rowkey с обновленной отметкой времени, которая должна быть обновлена ​​в таблице Кассандры.

Я пытался выбрать столбцы сразу после параметров, но, похоже, такого метода нет. 

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

Я использую Dataframe API.Таким образом, подход RDD не может быть использован.Как я могу это сделать?Cassandra версия 3.11.3, разъем Datastax 2.4.0-2.11

1 Ответ

0 голосов
/ 27 декабря 2018

Уточнение SaveMode используется для указания ожидаемого поведения сохранения DataFrame в источник данных (не только для c *, но и для любого источника данных).Доступны опции :

  1. SaveMode.ErrorIfExists
  2. SaveMode.Append
  3. SaveMode.Overwrite
  4. SaveMode.Игнорировать

В этом случае, поскольку у вас уже есть данные, и вы хотите добавить, вы должны использовать 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

...