问题标签 [hazelcast-jet]

在一个分区内，jet 是否按顺序处理每个项目，和/或是否有设置以这种方式进行配置？

谢谢香农

我正在尝试在 Spark Structured Streaming 和 Hazelcast Jet 之间映射一些概念，并了解其他主题。

Q1 - 在 Spark 中，每个 Kafka 分区将成为 spark 内部的一个分区，然后它们将由各个任务并行处理。我想我在某处读到 Hazelcast Jet 将合并来自 kafka 的所有消息，而不管 group.id 和主题分区，对吗？

Q2 - 我们如何增加 Jet 程序中的“消费者”数量以增加 kafka 消耗的吞吐量？在 Spark 中，我想我们只需要增加主题分区的数量就可以为新分区分配新的 spark 任务。

Q3 - 如果上面的 Q1 为真，是否可以避免合并和分发 kafka 分区以并行处理？一旦消息已经在 kafka 分区中分组和排序，合并所有消息意味着需要额外的处理来重新分区和再次排序消息。

Q4 - 如何定义每个顶点的数量？我的意思是，在字数统计示例中，我们有分词器和累加器，Jet 将如何决定/划分处理器的数量来创建分词器和累加器的实例？

这种新的编程范式对我来说非常新。我想用给定类中.map()的定义替换匿名函数。DistributedFunction但我不确定如何创建新功能。

我有以下管道：

基于一些 Jet 示例，我最终得到以下结果：

KafkaHelper 类：

DistributedFunction这种方法是否遵循将 a 传递给的规范/要求.map()？如果是，为什么？如果没有，我应该对其进行哪些更改？

我的用例需要读取来自 Kafka 主题的消息，并在消息发布到 Kafka 时按自然顺序处理这些消息。

Kafka生产者负责发布在单个kafka topic-partition中排序的每组消息，我需要在同一个Vertex-Processor中以相同的顺序处理每组消息。

上图代表基本思想。有一些从 Kafka 读取的 KafkaSource-Processors。

并且一条边连接到一个顶点以解码 kafka 消息等等。

我可以使用 kafka 消息键作为分区键，但我认为我最终会得到不平衡的解码处理器。

鉴于：

• 如何创建新的 Partitioner ？我找不到任何可以启发我的例子。
• 在新的分区器上，如何识别发出消息的 KS 处理器？我想在前一个顶点进程和下一个顶点处理器之间建立一对一的关系，例如，KS#0 总是将消息发送到 Decode#0，KS#1 到 Decode#1 等等。
• 我需要一个新的分区器还是有一些开箱即用的功能来实现它？

我是 Kafka 和数据流的新手，需要一些建议来满足以下要求，我们的系统预计每天有近 100 万条传入消息。该消息携带项目标识符。该消息应仅推送给该项目的用户。对于我们的案例，假设我们有项目 A、B 和 C。打开项目 A 的仪表板的用户只能看到/接收项目 A 的消息。

到目前为止，这是我对实现需求解决方案的想法，

消息应该在到达时被推送到 Kafka 主题，让我们将此主题称为根主题。消息一旦推送到根主题，可以由 Kafka 消费者/侦听器读取，并且基于消息中的项目标识符可以将该消息推送到项目特定主题。因此，任何消息都可以在主题 A 或 B 或 C 处结束。考虑使用 websockets 在消息到达项目用户的仪表板时对其进行更新。N 个项目主题将有 N 个消费者/听众。这些消费者会将项目特定的消息推送到项目特定的 websocket 端点。

请告知我是否可以对上述设计进行任何改进。

此处选择 Kafka 作为消息传递系统，因为它具有高度可扩展性和容错性。

在发送到客户端之前，没有复杂的转换或数据丰富。使用 Apache Flink 或 Hazelcast Jet 进行流式传输是否有意义，或者 Kafka 流式传输足以满足这个简单的要求。

另外，我什么时候应该考虑在我的项目中使用 Hazelcast Jet 或 Apache Flink。

当我必须在将消息发送给用户之前基于 Web 服务调用或数据库查找更新消息中的几个属性时，我应该使用 Flink 吗？

只有当我需要内存中的整个数据集来获得属性值时，我才应该使用 Hazelcast Jet吗？或者即使对于我上面指定的简单用例，使用 Jet 也会带来一些好处。请指教。

我正在尝试使用RoutingPolicy.ISOLATED在上游和下游之间创建专用连接，根据这个线程创建一个新的 Jet 自定义分区器

此外，尝试使用DiagnosticProcessors.peekOutputP来验证来自同一分区的消息是否正在发送到同一下游处理器。

在 Jet 初始化的日志下方：

我的 Kafka 生产者正在将具有相同密钥的消息路由到同一分区：

根据日志，看起来 kafkaSource 顶点正在将所有消息输出到同一个处理器：

在我的 Jet 代码下方：

更新

日志::解析

解码KafkaMessage

更新：2018-02-01

我正在为 KafkaSource 使用projectionFn，以便将 JSON 字符串解析为我的 kafka 消息值。

但是，kafka 文档将消息/记录定义为：

通过projectionFn，看起来我只能从每条消息中访问键和值。我怎么能通过Kafka timestamp.extractor访问时间戳值？

我正在尝试创建一个自定义包装器来记录处理每个顶点处理器执行的每条消息所花费的执行时间。

使用 DiagnosticProcessors.peekInputP/peekOutputP 和 processor.PeekWrappedP 作为灵感，我最终得到了以下代码：

和

现在我可以在我的应用程序顶点中使用我的包装器：

并得到预期的结果，

问题是包装器不适用于kafkaSource顶点。我试图了解 peekOutputP 背后的逻辑，但我无法获得源顶点的工作版本。

我应该怎么做才能让包装器也在源顶点中工作？

我正在尝试在https://blog.hazelcast.com/use-hazelcast-jet-stream-data-imap-kafka-topic/上执行为 Hazelcast Jet 提供的示例代码

但它失败了，但有一个例外

找不到名为“hz:impl:jetService”的服务！

这是我的堆栈跟踪

我在stackoverflow上遇到了一个类似的问题，但那是关于

找不到名为“hz:impl:cacheService”的服务

有人可以指导我找到解决方案，如何解决这个异常！

这个问题是关于 Hazelcast Jet 0.5.1 中的 Pipeline API

我正在尝试创建的管道有两个无限源：一个是代码（每分钟发送一个事件的自定义源），另一个是 Kafka 主题。

它看起来像这样：

当其中任何一个源发出事件时，我希望该事件经过相同的步骤并排入相同的接收器。如果我们将我的问题转换为 SQL 术语，我想要一个“UNION”。看起来像这样的东西：

目标管道

我发现的所有关于将两个节点合二为一的示例和文档都相当于 SQL“JOIN”操作，而不是“UNION”。

我发现绕过我的问题的唯一方法是做这样的事情，但我觉得这是框架应该已经具备的东西，尽管我似乎找不到它。

结果如下所示：

产生的管道