apache-spark - 使用目录引发 cassandra 连接器问题

Question

我正在按照说明操作 found here连接我的 spark 程序以从 Cassandra 读取数据。这是我配置 Spark 的方式:

val configBuilder = SparkSession.builder
  .config("spark.sql.extensions", "com.datastax.spark.connector.CassandraSparkExtensions")
  .config("spark.cassandra.connection.host", cassandraUrl)
  .config("spark.cassandra.connection.port", 9042)
  .config("spark.sql.catalog.myCatalogName", "com.datastax.spark.connector.datasource.CassandraCatalog")

根据文档，一旦完成，我应该能够像这样查询 Cassandra:
spark.sql("select * from myCatalogName.myKeyspace.myTable where myPartitionKey = something")
但是，当我这样做时，我收到以下错误消息:

mismatched input '.' expecting <EOF>(line 1, pos 43)

== SQL ==
select * from myCatalog.myKeyspace.myTable where myPartitionKey = something
----------------------------------^^^

当我尝试以下格式时，我成功地从 Cassandra 检索条目:

val frame = spark
  .read
  .format("org.apache.spark.sql.cassandra")
  .options(Map("keyspace" -> "myKeyspace", "table" -> "myTable"))
  .load()
  .filter(col("timestamp") > startDate && col("timestamp") < endDate)

但是，此查询需要执行全表扫描。该表包含几百万个条目，我更愿意利用谓词下推功能，这似乎只能通过 SQL API 获得。

我正在使用 spark-core_2.11:2.4.3、spark-cassandra-connector_2.11:2.5.0 和 Cassandra 3.11.6

谢谢!

Answer 1

Catalogs API 仅在尚未发布的 SCC 3.0 版中可用。它将随 Spark 3.0 版本一起发布，因此它在 SCC 2.5.0 中不可用。因此，对于 2.5.0，您需要使用 create or replace temporary view... 显式注册您的表。 , 如 described in docs :

spark.sql("""CREATE TEMPORARY VIEW myTable
     USING org.apache.spark.sql.cassandra
     OPTIONS (
     table "myTable",
     keyspace "myKeyspace",
     pushdown "true")""")

关于下推(它们对所有数据帧 API、SQL、Scala、Python 等的工作方式相同) - 当您的 timestamp 时会发生这种过滤。是第一个聚类列。即使在这种情况下，典型的问题是您可能会指定 startDate和 endDate作为字符串，而不是时间戳。您可以通过执行 frame.explain 来检查，并检查谓词是否被下推 - 它应该有 *谓词名称附近的标记。

例如，

val data = spark.read.cassandraFormat("sdtest", "test").load()
val filtered = data.filter("ts >= cast('2019-03-10T14:41:34.373+0000' as timestamp) AND ts <= cast('2019-03-10T19:01:56.316+0000' as timestamp)")
val not_filtered = data.filter("ts >= '2019-03-10T14:41:34.373+0000' AND ts <= '2019-03-10T19:01:56.316+0000'")

第一个 filter表达式将向下推谓词，而第二个( not_filtered )将需要完整扫描。