spark中cassandra行的写入时间答案

【问题标题】：writetime of cassandra row in sparkspark中cassandra行的写入时间
【发布时间】：2015-04-14 18:28:03
【问题描述】：

我正在使用带有 cassandra 的 spark，并且我想从我的 cassandra 表中选择我的行的 writeTime。这是我的要求：

   val lines = sc.cassandraTable[(String, String, String, Long)](CASSANDRA_SCHEMA, table).select("a", "b", "c", "writeTime(d)").count()

但它显示此错误：

java.io.IOException: Column channal not found in table test.mytable

我也试过这个请求

   val lines = sc.cassandraTable[(String, String, String, Long)](CASSANDRA_SCHEMA, table).select("a", "b", "c", WRITETIME("d")").count()

但它显示此错误：

<console>:25: error: not found: value WRITETIME

请问如何获得我的行的 writeTime。谢谢。

【问题讨论】：

标签： apache-spark cassandra-2.0

【解决方案1】：

编辑：此问题已在连接器的 1.2 版本中得到修复

目前连接器不支持从 Cassandra 读取时通过 CQL 函数。我已经注意到了这一点，并将为实现此功能开一张票。

https://datastax-oss.atlassian.net/browse/SPARKC-55

对于解决方法，您始终可以在操作中使用直接连接器，例如

import com.datastax.spark.connector.cql.CassandraConnector

val cc = CassandraConnector(sc.getConf)
val select = s"SELECT WRITETIME(userId) FROM cctest.users where userid=?"
val ids = sc.parallelize(1 to 10)
ids.flatMap(id =>
      cc.withSessionDo(session =>
        session.execute(select, id.toInt: java.lang.Integer)

代码修改自 Filter from Cassandra table by RDD values

【讨论】：

是否有可能并行化我的 ResultSet ？

【解决方案2】：

在cassandra-spark-connector 1.2中，可以通过写来获取TTL和写时间：

sc.cassandraTable(...).select("column1", WriteTime("column2"), TTL("column3"))

【讨论】：

我可以在 python 中做同样的事情吗？因为我收到了一个错误，因为 NameError: name 'WriteTime' is not defined

【解决方案3】：

看看this票。

有关用法，请查看集成测试here。

【讨论】：