用于joinWithCassandraTable的分区键检索

Question

我有以下卡桑德拉表格：

CREATE TABLE listener.snapshots_geohash 
(
    created_date text, -- date when record have come to the system
    geo_part text, -- few signs of geo hash - just for partitioning
    when timestamp, -- record creation date
    device_id text, -- id of device produced json data (see snapshot column)
    snapshot text, -- json data, should be aggregated by spark
    PRIMARY KEY ((created_date, geo_part), when, device_id)
)

每天早上，聚合应用程序应该加载前一天的数据，并从快照列聚合JSON数据。聚合将按地理哈希对数据进行分组，这就是为什么它的部分被选为分区键的一部分。

我知道它是有效的加载数据从Cassandra通过使用join的Cassandratable-但为此我必须得到RDD构造从（created_date，geo_part）对。虽然我知道created_date值，我不能列出geo_part值-因为它只是一部分的地理散列和它的值是不连续的。所以我必须以某种方式运行选择不同的created_date，geo_part从ks.snapshots和创建RDD从它的结果。问题是如何运行这个选择与火花2.0.2和cassandra-连接器2.0.0-M3或可能有其他方法？

Answer 1

val keys = sc.cassandraTable("listener","snapshots_geohash").select("created_date","geo_part").perPartitionLimit(1)

完整的解释见https://stackoverflow.com/a/56269424/17324。

Answer 2

我找到了通过使用CassandraConntor运行CQL查询来获取Cassandra分区密钥的方法：

 val cassandraConnector = CassandraConnector(spark.sparkContext.getConf)
 val distinctRows = cassandraConnector.withSessionDo(session => {
     session.execute(s"select distinct created_date, geo_part from ${keyspace}.$snapshots_table")
 }).all().map(row => {TableKeyM(row.getString("created_date"), row.getString("geo_part"))}).filter(k => {days.contains(k.created_date)})
 val data_x = spark.sparkContext.parallelize(distinctRows)

表结构设计有以下问题:Cassandra不允许添加created_date= ' ... '子句来选择不同的created_date、geo_part，并且需要获取完整的配对列表并在应用程序中进行过滤。

备选解决方案可以是使分区键连续。如果聚合将按小时进行，那么分区键可以是(创建日期，小时)，24小时可以在应用程序中列出。如果每天24个分区是不够的，并且聚合有group by by geohash，那么可以坚持使用geohash的重要部分，但是应该将其转换为可计数的部分，例如geo part . hash()% desirenumberofsubpartitions