从 Cassandra 查询的数据无法在同一列上再次过滤（InvalidQueryException）

Question

我正在尝试从 cassandra 按时间查询大量数据，然后使用 spark 数据集一次获取较小的数据块进行处理，但是，应用程序因无效查询异常而失败：

WARN  2018-11-22 13:16:54 org.apache.spark.scheduler.TaskSetManager: Lost task 0.0 in stage 2.0 (TID 5, 192.168.1.212, executor 0): java.io.IOException: Exception during preparation of SELECT "userid", "event_time", "value" FROM "user_1234"."data" WHERE token("userid") > ? AND token("userid") <= ? AND "event_time" >= ? AND "event_time" >= ? AND "event_time" <= ?   ALLOW FILTERING: More than one restriction was found for the start bound on event_time
        at com.datastax.spark.connector.rdd.CassandraTableScanRDD.createStatement(CassandraTableScanRDD.scala:323)
        at com.datastax.spark.connector.rdd.CassandraTableScanRDD.com$datastax$spark$connector$rdd$CassandraTableScanRDD$$fetchTokenRange(CassandraTableScanRDD.scala:339)
        at com.datastax.spark.connector.rdd.CassandraTableScanRDD$$anonfun.apply(CassandraTableScanRDD.scala:366)
        at com.datastax.spark.connector.rdd.CassandraTableScanRDD$$anonfun.apply(CassandraTableScanRDD.scala:366)
        at scala.collection.Iterator$$anon.nextCur(Iterator.scala:434)
        at scala.collection.Iterator$$anon.hasNext(Iterator.scala:440)
        at com.datastax.spark.connector.util.CountingIterator.hasNext(CountingIterator.scala:12)
        at scala.collection.Iterator$$anon.hasNext(Iterator.scala:408)
        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIterator.processNext(Unknown Source)
        at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
        at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$$anon.hasNext(WholeStageCodegenExec.scala:395)
        at org.apache.spark.sql.execution.SparkPlan$$anonfun.apply(SparkPlan.scala:234)
        at org.apache.spark.sql.execution.SparkPlan$$anonfun.apply(SparkPlan.scala:228)
        at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$$anonfun$apply.apply(RDD.scala:827)
        at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$$anonfun$apply.apply(RDD.scala:827)
        at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
        at org.apache.spark.scheduler.Task.run(Task.scala:109)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:338)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
Caused by: com.datastax.driver.core.exceptions.InvalidQueryException: More than one restriction was found for the start bound on event_time
        at com.datastax.driver.core.exceptions.InvalidQueryException.copy(InvalidQueryException.java:41)
        at com.datastax.driver.core.DriverThrowables.propagateCause(DriverThrowables.java:28)
        at com.datastax.driver.core.AbstractSession.prepare(AbstractSession.java:108)
        at com.datastax.driver.dse.DefaultDseSession.prepare(DefaultDseSession.java:278)
        at com.datastax.spark.connector.cql.PreparedStatementCache$.prepareStatement(PreparedStatementCache.scala:45)

这是我要执行的代码片段：

case class RawDataModel(userid: String, event_time: Long, value: Double)
var dtRangeEnd = System.currentTimeMillis()
var dtRangeStart = (dtRangeEnd - (60 * 60 * 1000).toLong)

val queryTimeRange = "SELECT * FROM user1234.datafile WHERE event_time >= " + dtRangeStart

val dataFrame = sparkSession.sql(queryTimeRange)

import sparkSession.implicits._
val dataSet: Dataset[RawDataModel] = dataFrame.as[RawDataModel]

dataSet.show(1)



dtRangeEnd = System.currentTimeMillis()
dtRangeStart = (dtRangeEnd - (15 * 60 * 1000).toLong)

val dtRangeData = dataSet.filter(dataSet("event_time").between(dtRangeStart, dtRangeEnd))

dtRangeData.show(1)

注意：这不是数据集问题，因为我已尝试将它们与数据帧交换，没有任何区别。一开始我认为这是一个懒惰的评估问题，同时懒惰地应用了两个不同的边界，但是 dataSet.show(1) 命令应该调用早期聚合并避免级联评估

Answer 1

Spark 将 sparkSession.sql(queryTimeRange) 和 dataSet.filter(dataSet("event_time").between(dtRangeStart, dtRangeEnd)) 合并为一个命令，在 cql 中如下所示：

SELECT "sensorid", "event_time", "value" FROM "company_5a819ee2522e572c8a16a43a"."data" WHERE token("sensorid") > ? AND token("sensorid") <= ? AND "event_time" >= ? AND "event_time" >= ? AND "event_time" <= ?

你在同一个字段上有两个相同的限制 "event_time" >= ?。

如果在执行 .filter 之前坚持 dataFrame Spark 将计算 dataFrame 与 .filter 分开：

val dataFrame = sparkSession.sql(queryTimeRange)
dataFrame.persist
dataFrame.filter(dataSet("event_time").between(dtRangeStart, dtRangeEnd))

从 Cassandra 查询的数据无法在同一列上再次过滤（InvalidQueryException）

Data queried from Cassandra cannot be filtered on same column again (InvalidQueryException)

scala

cassandra

datastax

apache-spark