用Spark在Cassandra中写RDD [实体]

Question

我正在尝试编写一个RDD，其中包含Cassandra中的公共类带有Spark，

class Test(private var id: String, private var randomNumber: Integer, private var lastUpdate: Instant) {
        def setId(id: String): Unit = { this.id = id }
        def getId: String = { this.id }
        def setLastUpdater(lastUpdater: Instant): Unit = { this.lastUpdater = lastUpdater }
        def getLastUpdater: Instant = { this.lastUpdater }
        def setRandomNumber(number: Integer): Unit = { this.randomNumber = randomNumber }
        def getRandomNumber: Integer = { this.randomNumber }
}

此类都有所有的设定器和获取器来维护封装，因此我不得不成为案例类，因为我必须在转换过程中修改值。

与卡桑德拉（Cassandra）中该实体相对应的表对应的表具有略有不同的字段名称：

CREATE TABLE IF NOT EXISTS test.test (
  id uuid,
  random_number int,
  last_update timestamp,
  PRIMARY KEY (id)
) 

我试图用方法savetocassandra编写此rdd，

implicit val connector = CassandraConnector(sc.getConf)
val rdd: RDD[Test]
rdd.saveToCassandra("test", "test")

但是该方法给我提供了属性名称巧合的例外在表格中带有字段名称的课程中，

Exception in thread "main" java.lang.IllegalArgumentException: requirement failed: Columns not found in entity.Test: [id, random_number, last_update]
at scala.Predef$.require(Predef.scala:277)
at com.datastax.spark.connector.mapper.DefaultColumnMapper.columnMapForWriting(DefaultColumnMapper.scala:106)
at com.datastax.spark.connector.mapper.MappedToGettableDataConverter$$anon$1.<init>(MappedToGettableDataConverter.scala:35)
at com.datastax.spark.connector.mapper.MappedToGettableDataConverter$.apply(MappedToGettableDataConverter.scala:26)
at com.datastax.spark.connector.writer.DefaultRowWriter.<init>(DefaultRowWriter.scala:16)
at com.datastax.spark.connector.writer.DefaultRowWriter$$anon$1.rowWriter(DefaultRowWriter.scala:30)
at com.datastax.spark.connector.writer.DefaultRowWriter$$anon$1.rowWriter(DefaultRowWriter.scala:28)
at com.datastax.spark.connector.writer.TableWriter$.apply(TableWriter.scala:433)
at com.datastax.spark.connector.writer.TableWriter$.apply(TableWriter.scala:417)
at com.datastax.spark.connector.RDDFunctions.saveToCassandra(RDDFunctions.scala:35)

我该如何在Cassandra 中编写实体，而不必将属性调用相同和 和属性是私有的 ？

Answer 1

savetocassandra允许您提供一个可选的colugnselector：

  def saveToCassandra(
    keyspaceName: String,
    tableName: String,
    columns: ColumnSelector = AllColumns,
    writeConf: WriteConf = WriteConf.fromSparkConf(sparkContext.getConf))(...): Unit

在您的情况下，您可以使用以下选择器：

def selector = SomeColumns(
  ColumnName("id"),
  ColumnName("random_number", alias = Some("randomNumber")),
  ColumnName("last_update", alias = Some("lastUpdate"))
) 

btw，而不是典型（建议）使用案例类，可以绝对将字段定义为var s，并从使用键入数据集中受益。这使得在写信给卡桑德拉之前将字段重命名非常容易。