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

Question

我有一个表模式

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

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

addedtime是聚类键

现在，当我在集群键 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 connector 1.4 中工作，但不适用于最新的 cassandra connector 1.6.0-M1。请让我知道问题

Answer 1

问题分析

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

做的时候

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

它将 addedTime 列转换为 String，然后进行比较。 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 中的一个错误，因为我们可能应该将谓词呈现给源，即使有一个“ Actor ”。不过，有一个解决方法，它涉及为 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)]