0

我正在编写一组扩展 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"))

...解决方案是将案例类定义移出“主要”功能(但我真的不知道这是否适用于上述问题......)。

4

1 回答 1

3

saveAsCassandraTable需要 3 个隐式参数。第一个 ( connector) 具有默认值,最后两个 (rwfcolumnMapper) 不在您的方法的隐式范围内saveResultsToCassandra,因此您的方法无法编译。

如果您需要有关隐式的更多信息,请查看另一个问题的答案。

saveResultsToCassandra如果您之前已经定义了表格 ( ),那么将您转换为下面的函数应该可以工作TableDef

def saveResultsToCassandra()(
  // implicit parameters as a separate list!
  implicit rwf: RowWriterFactory[T], 
  columnMapper: ColumnMapper[T]
) {
  rdd.saveAsCassandraTable("ks_name", "table_name")
}
于 2015-06-27T17:31:56.017 回答