问题标签 [hazelcast-jet]

我正在尝试运行 Hazelcast jet 0.5.1 代码示例。运行 Hazelcast 代码示例时出现以下异常：

\streaming\map-journal-source\src\main\java\MapJournalSource.java & \streaming\map-journal-source\src\main\java\RemoteMapJournalSource.java

日志详细信息如下。

你能帮我解决这个问题吗？

在 Jet 0.4 中，可以选择使用通用的group id.

在当前 Jet 0.5 中，此group.id值被忽略，我们如何防止 Jet 实例获取消息的重复副本？换句话说，我们希望工作负载（处理 Kafka 主题的消息）在 Jet 实例中进行负载均衡，我们如何使用 0.5 来做到这一点？

我们使用 Kafka 作为 Jet 工作的来源。我们希望将在 Kafka 中接收到的消息分发到在不同机器上运行的每个 Jet 实例。

我们可以使用 GroupConfig（设置名称和密码）和 JoinConfig（添加 IP 地址）成功地加入来自不同机器的成员。

当我们向 Kafka 源主题发送消息时，要么所有机器读取相同的消息，要么在一台机器上处理相同的消息两次。

例如，我们创建了一个 Kafka 主题，它有4 个分区和 2台不同的机器运行，正好有1 个喷气机集群。当我们运行应用程序并实例化 jet 实例时，我们看到2 个jet 成员已连接，并且分区在2之间拆分。

当发送像“message_one”这样的消息时，会发生以下两种情况之一：

1. 两个 Jet 实例都读取消息
2. 消息仅由 Jet 实例之一读取两次

在我们的情况下两者都不是必需的，因为我们不希望消息被复制。

如何防止两次处理消息？

下面是代码片段：

我正在尝试使用 Hazelcast Jet 0.5.1 运行服务器 - 客户端示例。

这个例子是从 hazelcast jet 下面的代码示例扩展而来的。

hazelcast-jet-0.5.1\code-samples\streaming\map-journal-source\src\main\java\RemoteMapJournalSource.java

这是一个场景

我有 1 个服务器程序和 2 个客户端程序。

服务器- 通常的 hazelcast 代码在 Hazelcast 映射中放置 100 个整数条目。它还为地图配置了事件日志。

客户端- 使用客户端配置连接到 hazelcast，从远程日志中读取条目并排放到 IList。

以下是观察结果

服务器将 100 个条目放入映射，但客户端 1 获得 200，客户端 2 获得 300 个条目（初始列表大小为 100）

我预计只有 100 个条目可用，并且条目将在这两个客户端之间分配。

示例代码如下。

服务器程序代码

客户 1 程序代码

客户端 2 程序代码。

你能帮我解决这个问题吗？

Hazelcast Jet 是否可以用于使用访问事件日志的多个客户端处理数百万条记录并且每个客户端将处理部分记录？

此外，是否可以累积不同客户端处理的结果？

我对 Hazelcast Jet 有以下疑问

用例如下

有一个应用程序（应用程序'A'，部署在集群中）使用 Hazelcast IMDG 并将数百万条记录/事务放在 hazelcast IMap 中。

已为此 IMap 配置事件日志。

还有另一个应用程序（应用程序 B，部署在集群中）实例化 JetInstance 并在每个节点上单独运行作业以处理记录。

目前，该作业从事件日志中读取数据并添加到 IList（参考 - hazelcast-jet-0.5.1\code-samples\streaming\map-journal-source\src\main\java\RemoteMapJournalSource.java）

由于作业在多个节点上运行，事件日志中的记录由多个节点处理。这会导致 IList 中有多个条目。

是否可以确保一条记录仅由“应用程序 B”的一个节点处理，而不由其他节点处理以避免重复？

如果不是，这是否意味着作业将由“应用程序 B”集群的单个节点运行？

这是一个示例代码（应用程序 B）

这是一个完整的代码。

应用程序源代码。

这是应用程序 B - 节点 1 示例代码

这是应用程序 B - 节点 2 示例代码

Hazelcast 客户端 - 将条目放入 Hazelcast 地图（应用程序 A）

以下是执行此操作的步骤。

1. 运行应用程序 A - Java 程序 RemoteMapJournalSourceSrv1

2. 运行应用程序 B 节点 1 - Java 程序 RemoteMapJournalSourceCL1

3. 运行应用程序 B 节点 2 - Java 程序 RemoteMapJournalSourceCL2

4. 为应用程序 A 运行 Hazelcast 客户端 - Java 程序 HZClient

此客户端程序根据控制台输入将条目放入映射中。请提供整数输入。

观察

执行时，.peek() 记录应用程序 B 的两个节点的值，并且在应用程序 A 映射中插入 1 个条目时，列表计数变为 2。

我有一个可能有空结果的滑动窗口和自定义聚合累加器。丢弃此类“空”聚合累加器进入接收器的正确方法是什么？

如您所见，聚合器返回字符串：

问题基本上是，在将可忽略的窗口/聚合结果与其他结果合并/扣除或冲入接收器之前丢弃它们的方法是什么？

嗨，我对 Hazelcast jet 有疑问。

我有一个应用程序在 , 中启动两个127.0.0.1:5701节点127.0.0.1:5702。当我运行应用程序时，它会创建 DAG 并获得所需的输出。如果我试图127.0.0.1:5703从另一个来源（而不是来自应用程序）连接 hazelcast jet 的另一个实例，则节点最初会加入但会引发序列化错误，例如

com.hazelcast.nio.serialization.HazelcastSerializationException：java.lang.ClassCastException：无法将 java.lang.invoke.SerializedLambda 的实例分配给字段 com.hazelcast.jet.core.processor.SourceProcessors。

所有配置均已启用 tcp-ip，因此我可以加入两个节点。

我还尝试先启动外部节点然后启动应用程序，反之亦然，两者都产生了相同的错误。我想知道为作业配置外部节点的正确方法是什么。喷气机版本0.6.1

提前致谢

当前的 Hazelcast Jet 0.6.1 代码示例演示了基于单个字段（例如ticker）的聚合。

这是一个参考。

\code-samples\streaming\stock-exchange\src\main\java\StockExchange.java

如何将其扩展到不止一个，例如ticker、traderId 等。

这是来自的当前示例代码 StockExchange.java

Hazelcast Jet 在控制台上打印 DAG 定义，一旦启动

这会将管道定义转换为 DAG。

这是一个管道定义。

这是打印在控制台上的 DAG 定义。

有没有办法获得包含所有细节的 DAG 定义，如滑动窗口细节、聚合 API 等？