Cassandra（+ Spark？）连续查询的最佳方法？答案

【问题标题】：Best approach to Cassandra (+ Spark?) for Continuous Queries?Cassandra（+ Spark？）连续查询的最佳方法？
【发布时间】：2015-03-01 11:31:16
【问题描述】：

我们目前使用 Hazelcast (http://hazelcast.org/) 作为分布式内存数据网格。这对我们来说效果不错，但是在我们的用例中，仅在内存中使用已经用尽了它的路径，我们正在考虑将我们的应用程序移植到 NoSQL 持久存储。经过通常的比较和评估后，我们几乎可以选择 Cassandra，最终选择 Spark 进行分析。

尽管如此，我们的架构需求存在差距，我们仍未掌握如何在 Cassandra 中解决（使用或不使用 Spark）：Hazelcast 允许我们在其中创建一个连续查询，只要添加/删除一行/从子句的结果集中修改，Hazelcast 用相应的通知回调。我们使用它通过 AJAX 流使用新的/更改的行不断更新客户端。

这可能是我们正在做的一个概念上的不匹配，所以 - 如何在 Cassandra 中最好地解决这个用例（无论有没有 Spark 的帮助）？ API 中是否有允许对键/子句更改进行连续查询的内容（尚未找到）？是否有其他方法可以获取密钥/子句更新流？某种事件？

我知道我们最终可以定期轮询 Cassandra，但在我们的用例中，客户可能对大量表子句通知感兴趣（想想“加利福尼亚海岸线上船舶位置的所有变化”），并且迭代出商店会破坏流媒体的可扩展性。

因此，神奇的问题是：我们缺少什么？ Cassandra 是不是适合这项工作的工具？我们是否不知道 apache 领域内/外的 API 或外部库的特定部分允许这样做？

非常感谢您的帮助！

雨果

【问题讨论】：

你检查过 Cassandra 触发器 (datastax.com/dev/blog/…) 吗？
嗨。谢谢。是的，我有 - 问题是触发器位于 Cassandra 的写入路径中，并且在那里放置通知/进一步写入可能会严重阻碍 Cassandra 众所周知的写入性能 - 我试图避免这些 :)

标签： events cassandra apache-spark cassandra-2.0

【解决方案1】：

我不是火花方面的专家，所以对此持保留态度，但也许你可以使用这样的方法：

使用 Spark Streaming 对传入数据流进行实时分析，并将位置更新实时推送给客户端。
使用 Cassandra 进行持久存储、缓存视图和数据汇总，客户端可以从中提取数据。

因此，您将编写一个连接到传入数据流的 spark 流应用程序，大概是一个定期报告船舶位置的应用程序。当它接收到船舶位置时，它会在 Cassandra 中查找船舶的最后一个已知位置（之前存储在该船舶 id 位置的集群时间序列中，按时间戳反向排序，以便最近的位置是第一行） .如果船舶位置发生了变化，spark 应用程序会在 Cassandra 中插入一个新的时间序列行并将新位置推送到您的实时客户端。

Spark 还将向 Cassandra 写入其他更新，以汇总客户可能想知道的内容，例如旧金山湾目前有多少艘船的表格。当客户端单击托架时，将查询汇总表以提取该数据以进行显示。任何需要客户端快速响应的东西都应该由 spark 预先计算并存储在 Cassandra 中以便快速检索。

当一个新客户端启动时，他们首先会查询（从）Cassandra 以获取所有船只的当前位置，然后会从 spark 应用程序推送对该数据的实时更新。

【讨论】：

【解决方案2】：

使用火花流。当需要更新时，执行两个操作：

执行 saveToCassandra，它将更新 cassandra 数据以供将来查询。
使用您正在使用的任何方式将更改推送到客户端。如果您的 AJAX 推送可以放入流式客户端代码中，您可以从 Spark Streaming 执行 AJAX 通知。否则，您可以向某个代理发送消息，该代理将中继到 Ajax 客户端。

您的代码可能如下所示：

val notifications = ssc.whateverSourceYouHaveThatGivesADstream(...)
notifications.foreachRDD(x => {
    x.foreachPartition(x => {
      cassandraConnector.withSessionDo(session => {
          x.foreach(y => {
              //use session to update cassandra
              // broadcast via AJAX or send to proxy to broadcast
          })
      })
    })
})

希望对您有所帮助。

【讨论】：

【解决方案3】：

Spark Cassandra Connector 可能会有所帮助。它支持从 cassandra 表进行流式传输：

import com.datastax.spark.connector.streaming._

val ssc = new StreamingContext(sparkConf, Seconds(1))

val rdd = ssc.cassandraTable("streaming_test", "key_value").select("key", "value").where("fu = ?", 3)

【讨论】：

【解决方案4】：

查看 Spark 作业服务器

你可能想看看Spark Job Server

它可以让你做一些事情，比如在不同的工作之间共享火花上下文——从而兑现 RDD。

并为 Spark 提供了一个RESTful API，用于近乎实时（基于刷新缓存的频率）查询。

【讨论】：