saveAsCassandraTable不是....dstream.DStream的成员[(String，Int，String)]

Question

当我试图使用saveAsCassandraTable将实时tweet加载到Cassandra时，我得到了以下错误

value saveAsCassandraTable is not a member of org.apache.spark.streaming.dstream.DStream[(String, Int, String)]

不过，我能够成功地使用saveToCassandra进行导出。到目前为止我能收集到的代码如下-

val tags = stream.map(_.getText).filter(_.startsWith("@xyz"))     
val Counts  =   tags.map((_, 1)).reduceByKeyAndWindow((x: Int, y: Int) => x + y, windowLength, slideInterval)    
val CountsAll   =   Counts.map{case (tag,   counter) => (tag,   counter, "Everything")}    
CountsAll.saveAsCassandraTable("demo1", "tags1")

我的要求是scala同时将tweet加载到Cassandra中的两个不同的表中。一个是Cassandra中的预定义表，另一个是每次加载tweet时动态创建一个表。这些tweet以2000秒的间隔加载。

感谢您对如何做saveAsCassandraTable的任何建议

Answer 1

就像@user6910411说的那样，您的错误告诉您正在尝试在DStream上调用DStream。您只能在RDD[T]上使用它，在您的例子中是RDD[(String, Int, String)]。

你想要这样的东西：

CountsAll.foreachRDD{ rdd =>
   rdd.saveAsCassandraTable("demo1", "tags1")
}

更新

RDD不支持截断等。您可能应该在Spark代码路径之外管理模式。但是，要直接解决您所要求的内容：

val conn: CassandraConnector = CassandraConnector(sparkConf())
conn.withSessionDo { session =>
      session.execute(s"""CREATE TABLE tags1..."")
      }
    }

其中，sparkConf()返回带有Cassandra设置的spark对象。.withSessionDo从DataStax C*驱动程序中为您提供了一个Session对象，因此您可以随意使用它。同样，我建议模式管理不是直接在Spark代码路径上完成，而是独立管理的。