我有一个kafka主题,它包含目录数据和以下命令:
项目\u插入
部分\u项\u更新
删除\u项
全部删除
现在我需要使用这个主题,可能流100k msgs/秒,一些数据库,这将有助于我把原始的命令流转换为项目流状态。所以数据库中只有当前项状态。基本上db将用于查找。
我的想法是:
在数据存储中插入/更新/删除项,
一旦处理了特定的消息,我将向另一个流发送新消息,告诉下游消费者某些项目已被插入/更新/删除。这些使用者随后将从数据存储读取项目的当前状态,并将项目状态摄取到另一个kafka主题。
我担心的是数据存储的酸性。它有多“酸”?它是否适合这样的用例?
我也在考虑使用更便宜的bigtable,但对于这个用例来说,这似乎不是正确的选择。
如果你对如何解决这个问题有什么想法/建议,我很高兴。
2条答案
按热度按时间fjnneemd1#
第一个问题是消息速率。数据存储不能维持超过1/秒的每个实体组写入速率(每个实体都是实体组的一部分),请参阅限制。因此,如果您期望每秒更新多个项/实体,则数据存储不适合。
要通过云数据存储实现acid,您需要避免最终的一致性。这是可能的。从读取实体值时的最终一致性来看:
通过使用仅键查询、祖先查询或按键查找(get()方法),可以避免读取实体值的最终一致性。我们将在下面更深入地讨论这些不同类型的查询。
我将放弃祖先查询作为一种可能性,因为它将要求所有相应的实体都位于同一实体组中,从而放大了上述写限制的影响。另请参见对单个实体组的更新。
棘手的是
upsert
操作,更具体地说是创建新实体和更新/删除现有实体之间的区别。如果不能始终从项数据生成/确定唯一的项标识符(或传递前一阶段确定的项标识符),则意味着您需要一个查询,该查询不能在事务内部执行,其结果将取决于最终的一致性。数据存储也不适合这种情况。
但是,如果您可以获得这样的唯一标识符,那么您可以将其用作实体键标识符,事情很简单:
upsert
操作变成了一个简单的事务性尝试get
实体通过该键(强一致性)和(在同一事务中):如果
get
失败,代码不存在,然后使用该键创建一个新实体如果
get
成功更新实体并将其保存回g52tjvyc2#
bigtable可以处理10节点集群的100k速率(我已经运行了多达3500个节点的测试,每秒处理35m更新)。bigtable对于单行具有很强的一致性
upserts
. bigtable用户设计的模式将他们所有的事务数据放在一行中。云bigtable支持
upserts
,并且没有区别insert
以及update
. 还有一个delete by range,理论上可以用于delete_all
案例。高交易率和低成本是使用cloudbigtable的正确理由。或者,您可以考虑使用cloud-panner,它用于高吞吐量事务数据。