我正在嘗試使用 Spark 在 Cassandra 中撰寫一個包含公共類的 RDD
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 }
}
這個類有所有的 Setter 和 Getter 來維護封裝，我需要它不是一個案例類，因為我必須在轉換程序中修改值。
Cassandra 中與該物體對應的表的欄位名稱略有不同：
CREATE TABLE IF NOT EXISTS test.test (
id uuid,
random_number int,
last_update timestamp,
PRIMARY KEY (id)
)
我正在嘗試使用該方法撰寫此 RDDsaveToCassandra
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 中撰寫物體而不必呼叫相同的屬性并且屬性在類中是私有的？
uj5u.com熱心網友回復：
saveToCassandra允許您提供可選的ColumnSelector：
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"))
)
順便說一句，雖然不是案例類的典型（和推薦）使用，但您絕對可以將欄位定義為vars 并從使用型別化資料集中受益。這使得在寫入 Cassandra 之前重命名欄位變得非常容易。
轉載請註明出處，本文鏈接：https://www.uj5u.com/gongcheng/497514.html