scala - saveToCassandra 基于来自 RDD 的内容

Question

我在scala中使用spark来构建一个通用应用程序来并行化http调用，我担心是否可以根据RDD的内容执行saveToCassandra操作，因为响应应该进入不同的表。

为了提供更多的清晰度，

val queries: List[Query] = List(Query("google", "fish"), Query("yahoo", "chicken"))
val inputRDD = sc.parallelize(queries)

在哪里

case class Query(dataSource: String, query: String)

然后将每个查询映射到要保存到 cassandra 的元组列表中，但根据查询中的数据源，google 的数据应该进入 google 的 cassandra 表，而 yahoo 的数据应该进入它自己的表。

TIA

Answer 1

我会简单地过滤并保存单个子集：

val keywords = Map("google" -> "googletab", "yahoo" -> "yahootab")
val keyspace: String = ???

val subsets = keywords.keys.map(k => 
  (k -> inputRDD.filter{case Query(x, _) => x == k}))

subsets.foreach{ case (k, rdd) =>
  rdd.saveToCassandra(keyspace, keywords(k), SomeColumns(???)) 
}