问题标签 [hazelcast-jet]

在 hazelcast Jet 中处理我的 DAG 时，我偶然发现了一个奇怪的问题。为了检查错误，我完全简化了我的方法，并且：根据教程，边缘似乎不起作用。

下面的代码几乎和它一样简单。两个顶点（一个源，一个汇），一个边。

源是从地图读取，接收器应该放入地图。

data.addEntryListener 正确地告诉我地图由另一个应用程序填充了 100 个列表（每个列表有 400 个字节的 25 个对象）......然后什么也没有。地图填满了，但 dag 根本不与之交互。

知道在哪里寻找问题吗？

我正在尝试将我几周前编写的 hazelcast Jet 0.3 DAG 系统重新设计为 v0.4，作为将其从批处理更改为流的第一步。有趣的是，突然间我遇到了一些奇怪的行为，我无法确定顶点是否按预期工作。试图了解正在发生的事情，我找不到如何窥视每个顶点内部工作的选项。有没有办法至少从中获取一些错误消息？

为了隔离问题，我试图将其简化为一个非常简单的“从列表读取，将其映射到映射写入映射”DAG。但是仍然没有成功。

在我愚蠢的例子下面，也许我犯了一个非常简单的错误，更有知识的人会马上看到？

出版商：

分析仪：

听众：

数据生成工作正常，一切正常，直到 DAG 接管......但没有错误消息或来自 DAG 的任何内容。有什么建议么？

我正在尝试使用 Hazelcast jet 分析传感器数据。为此，我需要打开一个 Tcp 套接字并处理通过它获得的信息。

我刚刚开始学习 Hazelcast Jet。我的来源是 UDP 数据报。我想在 Jet 的某些节点上并行处理它，并通过“域”将它们重新发送到其他地址。我想使用带有加载程序的 Hazelcast IMDG IMap 通过“源 ip”获取“域”。

但是当我尝试在 DistributedFunction 使用 IMap 时，我得到了异常

域映射器代码：

我的错误是什么？或者 Hazelcast Jet 对我的目的来说是错误的选择。

我在 Hazelcast Jet 使用 UDP 套接字作为数据源。

它仅在一个节点上正常工作。当我启动另一个节点（使用 hazelcase 多播连接）时，我的第一个节点尝试再次绑定 UDP 套接字。我还没有看到“关闭套接字”日志。

Hazelcast 日志在第一个节点：

七月 24, 2

更改集群成员后关闭 udp 套接字的正确方法是什么？

谁能说新的喷气机集群实例应该如何开始工作？

用例 1：

1. 通过 3 个节点启动 jet 集群
2. 提交作业到集群
3. 所有 3 个节点都开始作业并处理数据

用例 2：

1. 启动第 4 个节点
2. 第 4 个节点什么也不做，因为它不是新的提交作业命令

新的集群实例应该如何启动已经在另一个节点上启动的作业？

我有两张榛树地图 1.map2 2.map2

我想在钥匙上加入两张地图。

我可以在榛树里做吗？

谢谢

我WriteKafaP在 Jet 作业中使用顶点。发布到 Kafka 主题时，有没有办法插入自定义分区逻辑？

在运行分布式流之后，看起来这些快照没有从监控中删除或其他东西。

即使作业完成，这些只是继续每 5 秒记录一次。请告知是否需要有关实施的任何内容。

使用：

• 榛树 3.9.1
• 喷气机 0.5.1

2017-12-14 16:26:22,649 [hz.ttcore.cached.thread-8] DEBUG com.hazelcast.jet.impl.SnapshotRepository - [localhost]:18320 [ttcore] [0.5.1] Deleted all snapshots for job 857d-8e5e-0e20-df3b 2017-12-14 16:26:32,651 [hz.ttcore.cached.thread-1] DEBUG com.hazelcast.jet.impl.SnapshotRepository - [localhost]:18320 [ttcore] [0.5.1] Deleted all snapshots for job 857d-8e5e-0e20-df3b 2017-12-14 16:26:37,649 [hz.ttcore.cached.thread-1] DEBUG com.hazelcast.jet.impl.SnapshotRepository - [localhost]:18320 [ttcore] [0.5.1] Deleted all snapshots for job 857d-8e5e-0e20-df3b

作为 Hazelcast Jet 的新手，我试图建立一个设置，其中来自无限来源的单个项目（即用户请求的地图日志）针对（可能变化的和）巨大的参考项目地图进行 MapReduced。

具体来说，对于这个例子，我想在给定使用定义的输入向量（查询）的情况下确定向量映射（参考float[]）中最小欧几里德距离的向量（读取：）的 ID 。

如果在单台机器上天真地实现，这将遍历引用的 Map 项并确定每个项到查询的欧几里德距离，同时保持 k 最小匹配，其中输入来自用户请求（ HTTP POST、按钮单击等），计算完成后结果集立即可用。

我最近的方法是：

• 在地图日志上收听请求
• .distributed().broadcast()对映射作业的请求
• 让映射作业获得.localKeySet()参考向量的
• 发出 k 最小向量的 ID（按欧几里得距离）
• .partitioned(item -> item.requestId)通过分区减少/收集单个节点上的结果
• 将结果存储到客户端具有关键侦听器的地图中。

从概念上讲，这里的每个查询都是一批大小1，我实际上是在批量处理的时候。但是，我很难让映射器和化简器知道批次何时完成，以便收集器知道它们何时完成（以便他们可以发出最终结果）。

我尝试使用带有真实和虚假时间戳的水印（通过AtomicLong实例自动获得）并从tryProcessWm函数中发出，但这似乎是一个非常脆弱的解决方案，因为一些事件被丢弃了。我还需要确保没有两个请求是交错的（即在请求 ID 上使用分区），但同时让映射器在所有节点上运行......

我将如何攻击这个任务？

编辑＃1：

现在，我的映射器看起来像这样：

reducer 本质上是在复制它（当然，减去向量计算）。

编辑#2：

这是 DAG 设置。当前，当有多个并发请求时，它会失败。大多数项目由于水印而被丢弃。

编辑＃3：

这是我当前的组合器实现。我假设所有物品都会根据水印订购；或者一般来说，相同请求的项目将由相同的组合器实例收集。虽然这似乎不是真的......