通过sparklyr将cassandra表导入spark - 可以仅选择某些列吗？

我一直在与sparklyr要将大型 cassandra 表带入 Spark，请使用 R 注册它们并执行dplyr对他们进行操作。

我已经成功导入 cassandra 表，代码如下：

# import cassandra table into spark

cass_df <- sparklyr:::spark_data_read_generic(
  sc, "org.apache.spark.sql.cassandra", "format", 
  list(keyspace = "cass_keyspace", table = "cass_table")
  ) %>% 
  invoke("load")


# register table in R

cass_tbl <- sparklyr:::spark_partition_register_df(
         sc, cass_df, name = "cass_table", repartition = 0, memory = TRUE)
       )

其中一些 cassandra 表非常大（> 85 亿行），需要一段时间才能导入/注册，有些会导致内存溢出，即使有 6 个节点运行总共 60 个内核和 192GB RAM。但是，我通常只需要每个 cassandra 数据库中的一些列。

我的问题是：

1. 是否可以在导入/注册时过滤 cassandra 数据库，以便仅导入某些列，或者根据主键进行过滤（即通过传递SQL / CQL输入查询，例如SELECT name FROM cass_table WHERE id = 5)?
2. 这样的查询在上面的代码中会出现在哪里，语法采用什么形式？

我尝试将这样的查询添加为选项列表中的附加选项，即：

list(. . . , select = "id")

以及之前将其作为单独的管道调用%>% invoke("load"), i.e.:

invoke("option", "select", "id") %>%

# OR

invoke("option", "query", s"select id from cass_table") %>%

但这些都不起作用。有什么建议么？

您可以跳过急切缓存并选择感兴趣的列：

session <- spark_session(sc)

# Some columns to select
cols <- list("x", "y", "z")

cass_df <- session %>% 
  invoke("read") %>% 
  invoke("format", "org.apache.spark.sql.cassandra") %>% 
  invoke("options", as.environment(list(keyspace="test"))) %>% 
  invoke("load") %>% 
  # We use select(col: String, cols* String) so the first column
  # has to be used separately. If you want only one column the third argument
  # has to be an empty list 
  invoke("select", cols[[1]], cols[2:length(cols)]) %>%
  # Standard lazy cache if you need one
  invoke("cache")

如果您使用谓词可以显着减少获取的数据集量pushdown选项"true"（默认）并使用filter before缓存。

如果您想传递更复杂的查询，您可以注册临时视图并sql method:

session %>%
  invoke("read") %>% 
  ...
  invoke("load") %>% 
  invoke("createOrReplaceTempView", "some_name")

cass_df <- session %>% 
  invoke("sql", "SELECT id FROM some_name WHERE foo = 'bar'") %>%
  invoke("cache")