SparkSQL 下推过滤在 Spark Cassandra 连接器中不起作用

Question

我有一个表架构

appname text,
randomnum int,
addedtime timestamp,
shortuuid text,
assetname text,
brandname text,

PRIMARY KEY ((appname, randomnum), addedtime, shortuuid)

addedtime 是聚类键

现在当我在集群键添加时间上使用下推过滤器时，我没有看到它被应用

val rdd = tabledf.filter("addedtime > '" + _to + "'").explain
== Physical Plan ==
Filter (cast(addedtime#2 as string) > 2016-12-20 11:00:00)

根据文档，它应该被应用 https://github.com/datastax/spark-cassandra-connector/blob/master/doc/14_data_frames.md#pushdown-filter-examples

它还在 spark cassandra 连接器 1.4 中工作，但不适用于最新的 cassandra 连接器 1.6.0-M1。请告知问题

Answer 1

问题分析

问题似乎在于 Catalyst 处理比较的方式。

做的时候

val rdd = tabledf.filter("addedtime > '" + _to + "'").explain

它将 addedTime 列转换为字符串，然后进行比较。 Catalyst 不会将此谓词呈现给 Spark Cassandra 连接器，因此无法推送它。

INFO  2016-03-08 17:10:49,011 org.apache.spark.sql.cassandra.CassandraSourceRelation: Input Predicates: []
Filter (cast(addedtime#2 as string) > 2015-08-03)

这也是错误的，因为它正在做一个字符串比较（这在词法上可以在这里工作，但并不是你真正想要做的）所以这看起来像 Catalyst 中的一个错误，因为我们可能应该将谓词呈现给源，即使有一个“演员”。有一种解决方法，其中涉及为 Catalyst 优化器提供它想要看到的内容。

解决方法

如果我们给出类型提示

df.filter("addedtime > cast('2015-08-03' as timestamp)").explain

然后Spark会在没有字符串Cast的情况下生成正确的比较

DEBUG 2016-03-08 17:11:09,792 org.apache.spark.sql.cassandra.CassandraSourceRelation: Basic Rules Applied:
C* Filters: [GreaterThan(addedtime,2015-08-03 00:00:00.0)]
Spark Filters []

== Physical Plan ==
Scan org.apache.spark.sql.cassandra.CassandraSourceRelation@332464fe[appname#0,randomnum#1,addedtime#2,shortuuid#3] PushedFilters: [GreaterThan(addedtime,2015-08-03 00:00:00.0)]

Answer 2

你也可以使用 java.sql.Timestamp

val dateFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd")
val date = LocalDateTime.parse("2015-08-03", dateFormatter)
val timestamp= Timestamp.from(date.atZone(ZoneId.systemDefault()).toInstant)

df.filter($"addedtime" > timestamp).explain