3

我的拓扑有问题。我试图解释工作流程......我有一个每 2 分钟发出约 500k 元组的源，这些元组必须由一个 spout 读取并像单个对象一样处理一次（我认为是 trident 中的一批）。之后，螺栓/函数/还有什么？...必须附加时间戳并将元组保存到 Redis 中。

我尝试使用一个函数实现一个 Trident 拓扑，该函数使用 Jedis 对象（Java 的 Redis 库）将所有元组保存到 Redis 到这个 Function 类中，但是当我部署时，我在这个对象上收到一个 NotSerializable 异常。

我的问题是。如何实现在 Redis 这批元组上写入的函数？在网上阅读我找不到任何从函数写入 Redis 的示例或任何使用 Trident 中的 State 对象的示例（可能我必须使用它......）

我的简单拓扑：

TridentTopology topology = new TridentTopology();
topology.newStream("myStream", new mySpout()).each(new Fields("field1", "field2"), new myFunction("redis_ip", "6379"));

提前致谢

4

2 回答 2

7

（一般回复状态，因为与 Redis 相关的具体问题似乎在其他评论中已解决）

当我们牢记 Storm 从分布式（或“分区”）数据源（通过 Storm“spouts”）读取数据、并行处理多个节点上的数据流、可选地在这些节点上执行计算时，Storm 中数据库更新的概念变得更加清晰。数据流（称为“聚合”）并将结果保存到分布式数据存储（称为“状态”）。聚合是一个非常广泛的术语，仅表示“计算内容”：例如，在 Storm 中计算流上的最小值被视为先前已知的最小值与当前在集群的某个节点中处理的新值的聚合。

考虑到聚合和分区的概念，我们可以看一下 Storm 中允许在状态中保存某些内容的两个主要原语：partitionPersist 和 persistentAggregate，第一个在每个集群节点级别运行，无需与其他分区，感觉有点像通过 DAO 与数据库对话，而第二个涉及“重新分区”元组（即在集群中重新分配它们，通常沿着一些 groupby 逻辑），进行一些计算（“聚合” ) 在读取/保存某些内容到 DB 之前，感觉有点像与 HashMap 而不是 DB 交谈（在这种情况下，Storm 将 DB 称为“MapState”，如果地图中只有一个键，则称为“Snapshot”）。

要记住的另一件事是，Storm 的恰好一次语义不是通过只处理每个元组一次来实现的：这太脆弱了，因为我们的拓扑中定义的每个元组可能有多个读/写操作，我们希望避免出于可伸缩性原因的两阶段提交，并且在大规模下，网络分区变得更有可能。相反，Storm 通常会继续重放元组，直到他确定它们至少被完全成功处理过一次. 这与状态更新的重要关系是 Storm 为我们提供了允许幂等状态更新的原语 (OpaqueMap)，因此这些重放不会破坏先前存储的数据。例如，如果我们将数字 [1,2,3,4,5] 相加，则保存在 DB 中的结果始终为 15，即使由于某些原因在“sum”操作中多次重放和处理它们瞬态故障。OpaqueMap 对用于在 DB 中保存数据的格式有轻微影响。请注意，只有当我们告诉 Storm 这样做时，才会出现这些重放和不透明的逻辑，但我们通常会这样做。

如果您有兴趣阅读更多内容，我在此发布了 2 篇关于该主题的博客文章。

http://svendvanderveken.wordpress.com/2013/07/30/scalable-real-time-state-update-with-storm/

http://svendvanderveken.wordpress.com/2014/02/05/error-handling-in-storm-trident-topologies/

最后一件事：正如上面重放的内容所暗示的，Storm 本质上是一种非常异步的机制：我们通常有一些数据生产者在队列系统（例如 Kafka 或 0MQ）中发布事件，Storm 从那里读取。因此，按照问题中的建议从风暴中分配时间戳可能会或可能不会产生预期的效果：此时间戳将反映“最新的成功处理时间”，而不是数据摄取时间，当然它不会相同在重放元组的情况下。

于 2014-02-27T06:52:13.217 回答
1

您是否尝试过 redis 的 trident-state。github上有一个代码已经做到了： https ://github.com/kstyrc/trident-redis 。

让我知道这是否回答了您的问题。

于 2014-02-24T10:36:15.800 回答