问题标签 [hazelcast-jet]

我们计划使用快速聚合器或 hzc jet。我们需要基于无限的消息流进行聚合。因此，假设我收到相同类型的消息，它将缓存先前聚合的数据，或者它将根据该类型的所有先前消息进行计算，例如，假设我们必须聚合学生的标记和键是学生 ID，我们有 kafka 主题在哪里包含学生 ID 和分数以及其他一些细节的连续消息即将发布。现在假设我们收到学生 1 的 4 条消息，

1, 90 - 输出 90 2,80 - 170 3, 70 - 240 4 100- 370

在这种情况下，每次我们接收消息时，hazelcast jet 都会缓存最后的聚合结果，或者每次都会聚合所有消息

我正在尝试使用 hazelcast jet 对文件中的连续数字流求和

几个问题

1. 它没有给出预期的输出，我的期望是文件中出现新数字时，它应该将其添加到现有总和中
2. 为此，为什么我需要提供窗口和addTimestamp方法，我只需要做无限流的总和
3. 我们如何实现容错，即如果服务器重新启动，它将保存聚合结果，当它出现时，它将从最后计算的总和中聚合？
4. 如果服务器关闭并且当服务器启动时现在很少有数字进入文件，它会从服务器关闭时的最后一点读取还是会在服务器关闭时错过数字并且只会读取它在之后获得的数字服务器已启动。

我正在使用 hazelcast jet 来聚合（总和）数据流

源是 kafka，我在其中接收整数，喷射流只是添加每个传入的数字。

我有几个问题 1. 当它接收到每个数字时，它会将数据保存在 IMap 中，我如何访问该快照？

我有一个应用程序，我在其中创建了 Jet 实例和管道作业来聚合流数据的结果。我正在运行此类应用程序的多个实例。我面临的问题是因为有 2 个实例，它正在运行 2 个管道作业，因此计算结果两次并且不正确，但它发现两个 jet 实例都是同一个集群的一部分。

喷射管道是否不检查管道作业，如果相同，只需将其视为一个，就像卡夫卡流对其拓扑结构一样？

我们在服务器上托管了一个 Hazelcast 集群，并且在同一区域的不同服务器上的不同应用程序使用 Hazelcast Jet 客户端实例使用管道聚合来自 Kafka 源的数据。

在此设置中，Jet 客户端实例是否发送它从 Kafka 源 Hazelcast 集群接收到的数据，这将涉及大量 IO，或者当我们创建管道时，Hazelcast 集群本身会创建与 Kafka 的连接，并且此连接来自 Jet 集群而不是来自客户端应用程序？

我们有一个用例，我们从 kafka 接收需要聚合的消息。这必须以某种方式聚合，如果更新出现在相同的 id 上，那么如果需要减去现有值，则必须添加新值。

从各种论坛我了解到，jet 不存储原始值，而是汇总结果和一些内部数据。

在这种情况下，我该如何实现？

例子

我可以在每次添加时实现一个简单的使用。但是我无法实现需要减去现有值并添加新值的聚合。

1. 余额 1,2,3 是消息序列
2. 注释显示了 jet 执行的每条消息的聚合值。
3. 我的目标是添加新金额（如果 id 是第一次出现）并在更新的余额出现时减去金额，即 Id 与之前的相同。

我正在尝试使用 Jet 进行聚合，源和接收器是 Kafka 主题，要求是从源获取 GPB（google proto buf）消息并发布 GPB 消息。问题是我能够发布Double但不能发布 GPB 消息，它给了我编译错误。

这工作正常：

即使上面的代码工作正常，它也会发布到接收主题，而我的要求是发布一个具有接收主题属性double的 GPB 。当我尝试通过放一个beforedouble来做到这一点时，它给了我语法错误。以下是我尝试过的：mapdrainTo

金额是具有double属性的 GPB 消息。这给了我不理解的语法错误。你能帮我解决这个问题吗？

您能否分享一些文档或链接以及针对不同场景有不同聚合的地方？我浏览了 Hazelcast 示例、演示，但不是全部，但很少，但在那里没有找到我的用例。非常感谢。

关于如何在数据库表中插入新记录或更新现有记录时，Hazelcast jet JDBC 继续源获取数据？

我们正在对 kafka 与 jet 流进行 poc，我注意到 kafka 需要序列化器和反序列化器，但在 jet 流中没有看到它，所以只是想知道 jet 如何在集群之间和客户端之间发送数据。

另外，使用其中一种的优缺点是什么。

请注意，我们只比较流式处理部分，而不是 kafka 完全容错的部分，但 jet 并不像 jet 中的全部内存一样。

我们需要在流媒体部分找到更多的利弊。

实际上，在我们的架构中，我们使用Hazelcast IMDG，以便在多个服务器节点之间共享有关用户操作的信息。

我们的地图结构如下：[key:String|value: CustomObject].

现在，我们想要扩展我们的产品功能，并且我们想要开发一个实时仪表板，通过执行以下操作来执行实时数据流：

1. 复杂聚合
2. 连续查询
3. 等等

在流程结束时，我们希望将结果“发送”到Vert.x Eventbus，然后发送到套接字层 ( SockJS )，以便在仪表板中显示数据。

我们需要建立一个可扩展且快速的系统，以处理大量数据，例如每秒数千个事件。

第一张图片代表我们当前（旧）的架构，第二张图片代表我们的“目标”架构。

旧建筑

目标架构

您如何看待目标架构？

Hazelcast Jet的角色是否正确或是否有其他方式来执行这些操作（例如仅使用Hazelcast IMDG）？

提前致谢。