我正在使用Storm,它适用于许多用例。 最近,我看了一下Trident,它是Storm的高级抽象。 它支持仅一次处理,并使有状态的处理更加容易。 但现在我想知道... 为什么不能总是使用Trident而不是Storm呢? 我到目前为止阅读过的内容: Trident按批次处理消息,因此...
我正在为我的Trident拓扑实现一个IBackingMap,用于将元组存储到ElasticSearch中(我知道GitHub上已经存在几个Trident/ElasticSearch集成的实现,但我决定实现一个更适合我的任务的自定义实现)。 因此,我的实现是一个经典的带有工厂的实现:publ...
我是Trident in Storm的新手。我对TridentState感到困惑。就我所知,trident会为每个批次(无论批次中的所有元组是否通过在数据库中维护事务ID来完全处理)维护状态(即元数据),但我不确定以下语句的作用: TridentState urlToTweeters = ...
我正在学习Trident框架。在Trident的Streams上有几种方法可以对批处理中的元组进行聚合,包括this one,它允许使用Aggregator接口对元组执行有状态映射。但是,不幸的是,没有内置的相应功能来额外保留地图状态,就像其他9个重载的persistentAggregate(...
我对Apache Storm还不是很熟悉,一直在尝试使用Kafka的trident拓扑结构,即TransactionalTridentKafkaSpout。除了Storm UI之外,所有都运行良好。即使我没有向主题中生成任何数据,Storm UI仍然显示无效的发射/传输值。这意味着即使主题中没...
我找不到与正确将Kafka与Apache Storm Trident集成的相关文档。我试图查看这里以前发布的相关问题,但没有足够的信息。 我想将Trident连接到作为OpaqueTridentKafkaSpout的Kafka。这是目前正在运行的示例代码: GlobalPartitionI...
我们有一个Trident拓扑,使用kafka-storm OpaqueTridentKafkaSpout和TridentKafkaState同时从kafka消费消息并向其中生产消息。在我们的生产storm集群上运行一切正常,但在本地模式下运行时,我们经常遇到以下错误: java.lang.R...
根据Prometheus的storage.md,建议不要使用NFS存储作为Prometheus的持久卷。但是,像Prometheus Operator和OpenShift这样的解决方案展示了使用NFS作为Prometheus持久卷的示例。那么我错过了什么吗?如果不推荐使用NFS,那么为什么这些...
我能够使用标准的spout和bolt组合进行流式聚合,并且在正常情况下运行良好,当使用tick tuples在一定时间间隔内持久化数据以利用批处理时。目前我正在自己进行故障管理(跟踪未保存的元组等)。(即不是来自storm的ootb) 但我已经了解到Trident提供了更高层次的抽象和更好的...
我有一个拓扑结构,从 MQTT 代理接收数据,我希望一个 Spout 能够像这样工作: 每隔 x 秒发出一批元组(或者是单个元组中的字符串列表)。我该如何实现?我了解了一些 Storm Trident 的知识,但它的 IBatchSpout 似乎无法让我按照特定时间间隔以批量方式发出元组。...