问题标签 [trident]

我是风暴和三叉戟的新手。我写了一个 spout（使用 Storm 核心）来读取 RDBMS。我重写的open()方法BaseRichSpout

现在我想用 Trident 来做同样的事情，因为它内置了 join 功能。谁能告诉我，我们如何在 Trident 中实现相同的功能。我正计划实施IOpaquePartitionedTridentSpout。还有任何教程可以了解有关 Trident 的更多信息。

您如何强制中文“奇虎360浏览器”使用Chrome布局引擎而不是您的网站的三叉戟布局引擎。具体来说，我希望所有使用奇虎 360 浏览器访问我网站的用户都使用 Chrome38 引擎而不是 IE 引擎。

我创建了一个具有有效 html5 doctype 测试的站点，但它仍在尝试使用 trident 进行渲染。

注意：我已经让它在某些情况下工作。例如，如果我修改我的 /etc/hosts 127.0.0.1 taobao.com

然后在端口 80 上本地托管我的应用程序。如果我访问 taobao.com（重定向到本地主机），那么它使用 Chrome38。这意味着它正在使用某种域名白名单。

我正在为我的 Trident 拓扑实现 IBackingMap 以将元组存储到 ElasticSearch（我知道 GitHub 上已经存在几种 Trident/ElasticSearch 集成实现，但是我决定实现一个更适合我的任务的自定义实现）。

所以我的实现是一个带有工厂的经典实现：

您会看到它获取主机/端口/集群名称作为输入参数，并创建一个 ElasticSearch 客户端作为该类的成员，但它从不关闭客户端。

然后以非常熟悉的方式在拓扑中使用它：

这个拓扑被包装在一些公共静态 void main 中，打包在一个 jar 中并发送到 Storm 执行。

问题是，我应该担心关闭 ElasticSearch 连接还是 Storm 自己的事情？如果 Storm 没有做到这一点，在拓扑的生命周期中我应该如何以及何时做到这一点？

提前致谢！

我们如何CachedMap在 Trident 中使用。我的用例是这样的，我将做一些数据库，并希望将该信息存储在缓存或三叉戟状态中。下次当我得到任何元组时，我想在查询数据库之前查询三叉戟状态或缓存。我没有在三叉戟状态/缓存中获得所需的数据，我将查询数据库并更新三叉戟状态/缓存。请让我知道我们如何实现这一目标，或者是否有任何链接可以帮助我解决这种情况。

我们有一个 Trident 拓扑，它使用 kafka-storm OpaqueTridentKafkaSpout 和 TridentKafkaState 从 kafka 消费和生产。在我们的生产风暴集群上运行时一切正常，但在本地模式下运行时，我们经常收到以下错误：

我们还将此错误视为同一错误转储的一部分：

当前使用从github构建的最新storm 0.10.0-SNAPSHOT ，但是在稳定的0.9.3 版本中也出现了同样的问题。任何帮助将不胜感激，谢谢。

我正在尝试开始使用 Storm Trident 并设置拓扑并使用IOpaquePartitionedTridentSpout运行并由OpaqueMap支持。

但是，我很难找到让我的 spout/function 知道事务是否成功提交的方法。我没有在常规 Storm spout/bolt 接口中看到任何 ack 或 fail 方法。

我的用例是仅在同一类别的前一个被处理和持久化（或失败）时才发出一个类别的元组。因为然后我将使用处理后的数据来更新我的下一个类别的元组。来自不同类别的元组可以并行处理。

流按类别使用partitionBy方法进行分区。

设置max_spout_pending为 1 可以消除问题，因为 Trident 一次只提交 1 个批次。但这不是可扩展的。设置为大于 1 的任何值都会使相同类别的元组（如果它们在两个连续批次中发出）在前一个事务提交之前被处理。

或者我应该为每个类别设置一个喷嘴并将其设置max_spout_pending为 1？

谢谢

对于storm-starter中的一个简单的字数统计程序，逻辑相当简单：
1）将句子分成单词
2）发出每个单词
3）聚合计数（将计数存储在地图中）

但是这里有两个问题：
1）程序使用12个单独的线程来执行聚合部分，也就是说count不是GLOBAL，我们要多加一层才能得到全局count？
2）在bolt中，maps是用来存储count的，也就是说它有状态，如果当前worker失败了，bolt中存储的count都没有了怎么办？因为storm是无状态的
3）我们应该使用Trident来实现这个吗？

我想做的是按两个字段（"remote-client-ip", "request-params"）对流进行分组，并计算每个组中元组的数量。并将它们组合成一张地图。这是我的拓扑：

但是调试后发现数据流一开始就被阻塞了groupBy()，是多字段分组。Count()在随后的聚合语句中，我没有执行任何操作。

所以我想我误解了关于多字段分组和聚合之间交互的一些概念。

请让我知道我的猜测是对还是错。谢谢！

基本上，我想从 Couchbase 中读取一些带有元组属性的附加属性，例如，元组有输入字段“{a,b,c}”，我想发出一个元组“{a,b,c,d , e}" 其中 "d" 和 "e" 是从 Couchbase 中读取的，其中 "a" 作为键。

但是，Trident BaseFunction 是可序列化的，而 CouchbaseClient 不是，这会导致“NotSerializable”异常，有没有办法解决这个问题？我能想到两种可能的解决方案：

1）读取Trident外部的附加属性并将它们一起发送到Trident（我不想这样做因为我想把这种处理放在Storm中）

2）使用状态，只读（只实现multi-get，把multi-put留空，不知道这样行不行）

有什么简单的方法可以做到这一点？这在 Storm 本身中应该是可行的，并且在现实世界中很常见。它也适用于其他 NoSQL 数据库，例如 Cassandra。谢谢。

如何运行 Trident 拓扑的 JUnit 测试以允许元组在每个阶段测试和验证输出时流过拓扑？我曾尝试在 Storm 的测试框架中运行，但它无法允许对 Trident 进行验证和一致执行。

这是一个示例拓扑，其中包含一些我遇到最多问题的内嵌注释。

除此之外，我尝试编写自己的 BaseFilter 来标记存储所有元组的流的末尾，但似乎必须有更好的方法。此外，这并不能解决以受控方式运行拓扑的问题。这是三叉戟支持的东西吗？