问题标签 [apache-samza]

我有许多要运行的 Samza 作业。我可以让第一个运行正常。但是，第二个工作似乎处于 ACCEPTED 状态，并且在我杀死第一个工作之前永远不会转换到 RUNNING 状态。

这是 YARN UI 的视图：

以下是第二个作业的详细信息，您可以在其中看到没有分配节点：

我有 2 个数据节点，所以我应该能够运行多个作业。这是我的相关部分yarn-site.xml（文件中唯一的其他配置与 HA 配置、Zookeeper 等有关）：

编辑：

我可以在资源管理器日志中看到：

请问我做错了什么？

谁能告诉我 LinkedIn 使用 Kafka 的具体目的是什么。我从linkedin博客中阅读了很多关于Kafka的文章。他们在哪里解释了他们如何使用 Kafka 以及他们获得了多少性能优势。

Linkedin 是否使用 Kafka 通知网络中的其他用户您的朋友 xxx 有新的状态更新。类似的东西

有人让 Samza 在 HA 中与资源经理一起工作吗？如果是这样，我应该设置yarn.resourcemanager.hostname什么yarn-site.xml？

如果我将其设置为我的第一个 RM，如果我从该 RM 提交作业并且 RM 是活动的，则作业提交工作正常。如果我运行作业提交的 RM 机器未处于活动状态，我会在端口 8032 上收到连接被拒绝错误。

如果您使用 Samza 的OutgoingMessageEnvelope使用此格式发送消息：

并且您在流任务的 process() 方法中调用此方法并希望将传入消息路由到适当的分区，当您调用该方法时，Samza 会为您创建分区吗？

例如

如果我在流任务的process()wheremsg中调用消息实例：

这会自动为我创建分区 idA 和 idB（即，我是否需要在向它们发送消息之前创建这些分区）？我希望能够将消息路由到适当的分区，还希望能够使用单独的消息键记录压缩。

最近我正在尝试在 Samza 框架上做一些流处理工作。我已经成功部署了 hello-samza 示例。但是，当我尝试编写自己的工作时，我不知道从哪里开始工作。

我已阅读此文档，但我仍然无法理解重点。所以任何人都可以帮助我：

1. 我的代码的架构是什么（源代码、lib 代码和配置）。
2. 我的代码将推送到哪个目录。
3. 我需要做哪些其他工作才能让我的代码运行。

你的建议对我很有帮助，非常感谢！

我正在为我们正在我的公司构建的新事件和流处理系统提出一组模式，以将几个当前断开连接的系统连接在一起。我们已经明确定义了 12 个领域模型，现在正尝试将所有应用程序将事件发送到融合 (kafka) 平台的一组事件模式组合在一起。然后这些将由 samza 提取和处理，以执行各种工作，然后为我们的特定领域服务填充数据库。

这一切都很好，我们从每个域的一个事件开始（例如地址）但是，我们很快遇到了需要为不同类型的事件提供不同数据的问题。例如，创建地址的事件需要域中的所有（或大部分）字段。而更新只需要一个 id 和正在更新的内容。

那么，我正在寻找的是那些过去做过这件事的人的一些建议？理想情况下，我希望每个域只使用一个事件模式来保持干净。这样，我们每个事件都有一个对应的 kafka 队列，可以轻松地重播以重新获得状态或返回到特定的先前状态。然而，感觉更简单、更实用的方法是为每个动词使用单独的模式（即创建、更新、删除）

一些相关的堆栈细节：

Confluent REST 代理 -> avro -> kafka -> samza -> 各种数据库。

我使用 Apache 作为 Kafka 的 Samza 框架，我需要在 2 个任务之间共享相同的 RockDB 键值存储。

我可以在没有存储键值并发的情况下这样做吗？

现在我做了 samza 的基本设置。之后，我处理来自 kafka 代理主题的 1,00,000 条 json 记录，每条记录大小为 100 字节，但每秒我只能处理 11000 条记录。如何提高处理速度？

使用以下配置，我可以将 samza 连接到 kafka-broker

但是我对 SystemFactory 类有一些疑问。如何编写我们自己的systemfactory类？SystemFactoryClass 的目的是什么？请给我一些想法

我正在测试和调试一个在 kafka 之上运行并使用 samza 的事件源（或有状态的流处理）应用程序。我想删除 kafka 中的队列和主题，以便 samza 作业在启动时获得一个空的 kafka 安装。

我该怎么做？？

编辑：

这个问题比我最初写的要复杂和具体。

正如大卫所说，有一个清除主题的解决方案，从 kafka 0.8.2 开始： Purge Kafka Queue

我感兴趣的是设置一个自动加载 zookeeper 和 kafka 的测试环境（它们作为二进制包捆绑在我的 git 存储库中）。

我正在使用 gradle 环境 + eclipse 和 JUnit。我从 Eclipse 运行集成测试（作为 JUnit 测试）。

加载如何实现自动化？我应该创建一个特定的测试类来设置环境并启动 kafka 和 zookeeper 吗？有没有参考示例/代码？这个想法是加载环境，运行一些测试，然后停止。如果这个过程可以在几秒钟内完成，那就更好了。