Spark：如何在保存到cassandra时配置writetime

Question

我有一个类似于卡桑德拉表的实体。我正在使用火花将数据保存/更新到卡桑德拉中。这里的实体是提供案例类

case class Offer(offer_id: String, metadata_last_modified_source_time: Timestamp, product_type: String, writeTime: util.Date)

val offerDataset: Dataset[Offer] = ....

我将这些数据保存如下

offerDataset.write.format("org.apache.spark.sql.cassandra")
      .options(Map("keyspace" -> cassandraKeyspace, "table" -> tableName))
      .mode(SaveMode.Append)
      .save()

cassandra表的模式是

OFFER(offer_id, metadata_last_modified_source_time, product_type) 

问题是在保存/更新cassandra表时，将Offer实体的writeTime字段配置为写入时间戳。这是在《税务》中提到的-https://github.com/datastax/spark-cassandra-connector/blob/master/doc/reference.md配置

writetime=columnName

我不明白的是语法应该是什么样子。

任何帮助都将不胜感激

Answer 1

本文档适用于Spark Cassandra连接器的alpha版本，因此请期待出现一些不起作用的情况。正如文档中所指出的，这是一个表格选项，因此您可以通过选项设置它。只需从util切换即可。日期到时间戳或长类型-Spark SQL不支持从日期类型进行编码。

使用以下定义一切工作：

import java.time.Instant
import java.sql.Timestamp

case class Offer(offer_id: String, metadata_last_modified_source_time: Timestamp, 
  product_type: String, writeTime: Long)

val offerDataset = Seq(Offer("123", Timestamp.from(Instant.now()), "test", 1243124234L),
  Offer("456", Timestamp.from(Instant.now()), "test", 12431242366L)).toDF

或者使用Timestamp：

case class Offer(offer_id: String, metadata_last_modified_source_time: Timestamp, 
   product_type: String, writeTime: Timestamp)

val offerDataset = Seq(Offer("123", Timestamp.from(Instant.now()), "test", new Timestamp(1243124234L)),
  Offer("456", Timestamp.from(Instant.now()), "test", new Timestamp(12431242366L))).toDF

如果我们使用以下表格结构：

create table test.wrt_test (
  offer_id text,
  metadata_last_modified_source_time timestamp,
  product_type text,
  primary key(offer_id, metadata_last_modified_source_time));

然后可以将数据保存为以下格式（仅在3.0-alpha中！）：

import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.cassandra._
offerDataset.write.cassandraFormat("wrt_test", "test")
    .option("writetime", "writeTime") // here you specify name of the column with time!
    .mode(SaveMode.Append).save()

但是，如果您使用RDD API，它在当前版本中也可以正常工作：

import com.datastax.spark.connector.writer._
offerDataset.rdd.saveToCassandra("test", "wrt_test", 
   writeConf = WriteConf(timestamp = TimestampOption.perRow("writeTime")))

在这两种情况下，你都会得到以下结果：

cqlsh> select offer_id, metadata_last_modified_source_time, product_type, writetime(product_type) from test.wrt_test;
offer_id | metadata_last_modified_source_time | product_type | writetime(product_type)
----------+------------------------------------+--------------+-------------------------
      123 |    2020-04-16 07:28:38.905000+0000 |         test |              1243124234
      456 |    2020-04-16 07:28:38.905000+0000 |         test |             12431242366
(2 rows)