我正在编写一组扩展 Spark RDD API 的方法。我必须实现一种存储 RDD 的通用方法,首先我尝试包装 spark-cassandra-connector 的saveAsCassandraTable
,但没有成功。
这是“扩展 RDD 的 API”部分:
object NewRDDFunctions {
implicit def addStorageFunctions[T](rdd: RDD[T]):
RDDStorageFunctions[T] = new RDDStorageFunctions(rdd)
}
class RDDStorageFunctions[T](rdd: RDD[T]) {
def saveResultsToCassandra() {
rdd.saveAsCassandraTable("ks_name", "table_name") // this line produces errors!
}
}
...并将对象导入为:import ...NewRDDFunctions._
.
标记的行会产生以下错误:
Error:(99, 29) could not find implicit value for parameter rwf: com.datastax.spark.connector.writer.RowWriterFactory[T]
rdd.saveAsCassandraTable("ks_name", "table_name")
^
Error:(99, 29) not enough arguments for method saveAsCassandraTable: (implicit connector: com.datastax.spark.connector.cql.CassandraConnector, implicit rwf: com.datastax.spark.connector.writer.RowWriterFactory[T], implicit columnMapper: com.datastax.spark.connector.mapper.ColumnMapper[T])Unit.
Unspecified value parameters rwf, columnMapper.
rdd.saveAsCassandraTable("ks_name", "table_name")
^
我不明白为什么这不起作用,因为saveAsCassandraTable
它设计用于任何 RDD。有什么建议么?
我对spark-cassandra-connector 文档中的示例有类似的问题:
case class WordCount(word: String, count: Long)
val collection = sc.parallelize(Seq(WordCount("dog", 50), WordCount("cow", 60)))
collection.saveAsCassandraTable("test", "words_new", SomeColumns("word", "count"))
...解决方案是将案例类定义移出“主要”功能(但我真的不知道这是否适用于上述问题......)。