问题标签 [hazelcast-jet]

我正在使用 hazelcast jet 对流数据执行聚合。问题是，hazelcast 客户端意外关闭。

我已经用远程地图源实现了简单的管道，然后结果就被简单地沉没了。

在客户端，第一个 cca 的输出与预期一致。30秒。然后关闭发生，然后，这些打印值冻结。好的，这是合乎逻辑的，因为它已被关闭。但是，如何防止关机？

在服务器端，它说连接被另一端关闭 - 所以，我的客户端：

我很乐意得到一些方向和想法，在哪里寻找问题。

我收到“事件已删除，晚了 5051 毫秒。”

我应该如何构建处理所有事件的管道，无论它们是否迟到。

我尝试了几种方法。基本上，我尝试的是

1. 如果没有窗口，我没有收到迟到的事件，但这不适用于我，因为并行执行和接收器中的值被覆盖而不是合并。
2. 因此我使用了窗口，它解决了我最重要的问题，但导致了迟到的事件。
3. 接下来，我尝试使用不带时间戳的窗口，这引发了必须定义时间戳的异常。

基本上我在这里有两个问题：1）如何将新事件合并到接收器中的现有事件 2）而不丢弃事件或覆盖。

代码：

我有一个在 AWS EC2 实例上运行的分布式系统。我的集群有大约 2000 个节点。我想介绍一个流处理模型，它可以处理每个节点定期发布的元数据（cpu 使用情况、内存使用情况、IO 等）。我的系统只关心最新数据。当处理模型关闭时，丢失几个数据点也是可以的。因此，我选择了 hazelcast-jet，它是一种性能出色的内存处理模型。在这里，我有几个关于模型的问题：

1. 将 hazelcast-jet 部署到多个 ec2 实例的最佳方法是什么？
2. 如何从数千个来源中提取数据？源推送数据而不是被拉取。
3. 如何配置客户端以便它知道在哪里提交任务？

如果有一个我可以学习的综合示例，那将非常有用。

我按照此处的说明在 aws ec2 上设置了 Hazelcast-jet 集群。我使用了 hazelcast-aws 模型，以便节点可以自动发现彼此。集群已启动并运行：

它还成功地找到了它的对等点：

除了我的客户端（不在 aws 上）无法连接到集群之外，一切看起来都很好。我所做的只是运行单词计数示例。唯一的区别是，我不想让客户端和服务器都在同一个 JVM 中运行，而是将任务提交到我设置的集群。我JetInstance jet = Jet.newJetInstance();用（按照说明）替换了：

我可以告诉客户正在寻找正确的端点：

我已经将 5701 添加到两个 ec2 实例使用的安全组的入站规则中。为了调试，我运行了几个网络命令来查看端口 5701 是否打开：

我对网络的了解有限。我无法弄清楚问题是什么。我注意到的一件事是，当客户端尝试连接到私有 ipv4 地址时，为 ipv6 打开了端口。

在处理流后（应用过滤器等），处理后的流中的每个剩余项目是否可以直接触发事件而不是写入接收器？

我有一堆设备不断向 hazelcast 喷射流服务器报告状态。服务器对流执行一些过滤，只保留真正状况不佳的设备。我希望这些设备直接触发一些维护事件，而不是被写入接收器（例如 IMAP）。目前，我正在做的是将流接收到 IMAP 并在地图上注册侦听器。但是，我不确定这是否是我用例的最佳实践。我主要有两个担心：

1. 我不一定需要存储这些设备。我需要的只是告诉其他服务该设备需要维护。
2. Doc 说不应该把沉重的逻辑放在听众身上。事件触发的任务比较全面怎么办？

我正在关注Doc以尝试如何通过直接从 IMap 中查找来丰富无界流。我有两个地图：

1. 产品：Map<String, Product>（ProductId 作为键）
2. 卖家：（Map<String, Seller>以SellerId为key）

Product和都是Seller非常简单的类：

我有两个数据生成器不断将数据推送到两个地图。两个地图都启用了事件日志。我已经验证事件日志工作正常。

我想Product用Seller地图丰富地图的流事件。这是我的代码片段：

提交作业时，出现以下错误：

com.hazelcast.spi.impl.operationservice.impl.Invocation - [172.31.33.212]:80 [jet] [3.1] 异步执行回调失败：com.hazelcast.util.executor.DelegatingFuture$DelegatingExecutionCallback@77ac0407for call Invocation{ op=com.hazelcast.map.impl.operation.GetOperation{serviceName='hz:impl:mapService', identityHash=1939050026, partitionId=70,replicaIndex=0, callId=-37944, invocationTime=1570410704479 (2019-10-07 01:11:44.479), waitTimeout=-1, callTimeout=60000, name=sellerMap}, tryCount=250, tryPauseMillis=500, invokeCount=1, callTimeoutMillis=60000, firstInvocationTimeMs=1570410704479, firstInvocationTime='2019-10-07 01 :11:44.479', lastHeartbeatMillis=0, lastHeartbeatTime='1970-01-01 00:00:00.000', target=[172.31.33.212]:80, pendingResponse={VOID}, backupsAcksExpected=0, backupsAcksReceived=0,连接=空}

我试图在我的集群中放置一个和两个实例并得到相同的错误消息。我无法弄清楚根本原因是什么。

我找不到从 MultiMap 构造 Source 的任何可能性，为什么标准 API 中没有提供这样的 Source？ https://docs.hazelcast.org/docs/jet/3.2/manual/#overview-of-sources-and-sinks

想象一个处理系统，它从其他内部系统或从互联网公开的 REST 端点接收数据。收到后，一组（后台）任务（根据数据库或其他 Web 服务计算来自输入的响应）并行处理数据，一些连续的一个接一个，或完成少数并行任务的一些任务。最终（边缘）任务可以将结果发送到外部 Web 服务或存入数据库表。

这些任务可以设计为一个简单的服务，处理message-queues由其他此类服务传递给它的消息，并将结果发送到另一个队列以进行进一步处理。系统可以使用流来处理，而不是队列。我发现这种方法（编写由消息代理连接的服务）必须重复处理它们的故障场景、扩展和吞吐量。

是否有任何特定的框架（如 Flink 或 Samza 或 Storm 或 Airflow 或 Siddhi 或 Hazelcast Jet）来实现这些任务并将它们部署到一个负责扩展和故障处理的集群中，而不是将每个任务实现为详细的服务（例如。 ，在春天）并独立缩放它们。

我正在尝试在分布式系统上运行 Hazelcast-jet示例之一。我的目标是在分布式系统上运行代码，利用多台机器的处理能力。我有两台笔记本电脑通过 LAN 连接。当我在一台机器上运行此示例时，它运行良好，要在两个系统中运行它，我只使用 jet 实例启动机器 1 和机器 2。两台机器上的代码都是，

机器 1

机器 2

错误

当我在单台机器上运行此代码时，Example 工作正常并且输出符合预期。

谁能帮我解决我做错了什么？

我在 Kubernetes 集群（Openshift）中有一个 Jet（3 实例）集群设置。我想向它提交工作。我使用代理连接但未提交作业。我还尝试将远程 jar 文件上传到其中一个集群 pod 并从那里提交 - 不幸的是，我无法连接到任何集群。

编辑：其他集群详细信息

我使用端口转发连接到 Jet 集群

oc port-forward --namespace jet svc/jet-cluster-hazelcast-jet 5701:5701

然后，如果我从我的系统发出jet.sh cluster命令，我将无法连接到任何 Jet 集群错误。

然后提交我使用的作业，

jet.sh submit -v target/job1-1.0-SNAPSHOT-jar-with-dependencies.jar

错误

任何文档、示例都会有很大帮助。