问题标签 [trident]

我对 Apache Storm 很陌生，并且一直在尝试使用 Kafka 的三叉戟拓扑，即TransactionalTridentKafkaSpout。除 Storm UI 外，一切正常。即使我没有为我的主题生成任何数据，Storm UI 仍然显示无效的发出/传输值。这意味着即使主题中没有数据，计数也会不断增加。我已经尝试删除存储在 zookeeper、storm、kafka 中的数据/日志并重新创建 kafka 主题，并且还设置了

但问题仍然存在。

我还发现了一个名为Capillary的工具来监控风暴集群。我正在使用以下属性

我在这里使用 Kafka 的嵌入式 zookeeper。即使这也无法获得以下异常。

对任何一个的任何帮助都会很棒。提前致谢。

配置和源代码片段：

拓扑摘要图像：

我写了一个关于三叉戟的演示。在演示中，我想在事务喷口中生成一系列批次的数据，然后对它们进行持久聚合操作。

正常处理是这样的：

step1: txid=1, {"aaa 3", "bbb 2"}==>persist to DB(OK)

step2: txid=2, {"ccc 6", "ddd 7"}==>persist to DB(OK)

在step1中persist to DB的操作出现异常的情况下，我想流程应该是

step1: txid=1, {"aaa 3", "bbb 2"}==>persist to DB(NG)

step2: txid=1, {"aaa 3", "bbb 2"}==>persist to DB(OK)

但测试结果如下：

step1: txid=1, {"aaa 3", "bbb 2"}==>persist to DB(NG)

step2: txid=1, {"ccc 6", "ddd 7"}==>persist to DB(OK)

我认为这个过程是不正确的。但我不知道为什么会这样。我检查了动物园管理员，在路径“meta/1”中，值为空。

我自己写了一个事务性的spout，实现了接口ITridentSpout，我认为这可能是我写的spout的问题。

任何人都可以给我一些建议吗？或者给我一个实现的事务喷口示例供我参考。

谢谢！

我构建了一个 topo 以通过 kafka 获取消息，然后 grep 一些关键字，如果合适，写入本地文件。

我使用OpaqueTridentKafkaSpoutstorm-kafka来保证元组不会丢失或重复，但是考虑一种情况：将消息写入本地文件时，发生一些错误（例如，空间不足）。此时，有些消息已写入本地文件，有些则没有，如果 spout 重新发送消息，则消息将被写入两次。

如何处理？

从 这个文档链接中，我可以了解到使用以下命令很容易重新平衡拓扑

但是如何重新平衡三叉戟拓扑，因为在三叉戟拓扑中我们不能用名称标记过滤器/功能（螺栓），那么如何增加/减少并行计数？或者有没有办法使用特定拓扑中所有组件的乘数来增加/减少并行计数？

我从不同的 Kafka 主题中读取了多个 OpaqueTridentKafkaSpout。我希望所有这些流中的数据通过同一组函数。实现这一目标的最佳方法是什么。我是否需要创建单独的流并将每个元组再次传递给同一组函数。像下面这样？

I am writing a Trident topology to process stream of data from Kafka and feed in to Redis and Cassandra. I am able to write the data in to Cassandra. Now I would like to write the same data in to Redis.

Is there a way to duplicate the tuples and branch it in to 2 flow where one goes in to Redis and another goes in to Cassandra?

Core Storm 允许通过可插入的调度程序指定在哪个节点上执行特定的 Bolt。我们正在考虑将 Storm 拓扑重新设计为 Trident，但没有看到我们如何获得相同的行为，即在哪些节点上配置一些特定的螺栓应该使用 Trident 拓扑执行。Trident 是否仍可使用日程安排？谢谢你。

我是三叉戟的新手。我正在编写一个从 kafka 读取数据的三叉戟拓扑。主题名称是“测试”。我有本地卡夫卡设置。我在本地启动了 zookeeper，kafka。并在 kafka 中创建了一个主题“测试”并打开了生产者并输入了消息“Hello Kafka！”。

我想使用 trident 从“测试”主题中读取消息“Hello Kafka”。

下面是我的代码。我得到空元组。

这是我的 TestFilter 类代码

每当我在 kafka 生产者中向“测试”主题键入消息时，首先打印 sysout，但它没有通过 if 循环。我只是收到消息'TestFilter 被调用...'仅此而已。

我想将我生成的实际数据用于“测试”主题。如何？

我想构建一个三叉戟拓扑，它将从 Kafka 获取元组，按一个字段对它们进行分组，并将整个分组的元组保存在 HBase 中。分组的原因是，与放置单个对象相比，使用多个放置对象的 HBase 放置操作更快，我想Put按一个字段对对象进行分组并将所有这些Put对象插入单个 HBase API 调用中。

我已经编写了以下代码，但不确定我应该在哪里编写将所有元组转换为 HBase 'Put' 类型对象并将所有这些 Put 对象保留在 HBase 中的代码。

谁能帮帮我吗？

我是 Storm 的新手，基本问题很少。我的storm用例是流处理和批处理。

用例 #1：Storm 拓扑将元组作为流接收并对其进行处理。

用例 #2：Storm 拓扑应该将元组作为一批元组接收并对其进行处理。

我使用 Kafka 作为队列机制来提供 Storm 拓扑。

问题：有没有办法，我可以告诉我一个特定的元组是流的结尾，风暴应该告诉我所有元组的处理何时完成？

Storm 不是执行此操作的正确框架，因为它用于流处理（用例 #1）。Storm Trident 对用例 #2 有帮助吗？