1

我对 Kafka 和 Samza 很陌生。我尝试了 hello-samza 示例，它正在工作。我正在寻找的是创建一个 samza 任务，该任务从 kafka 主题中读取消息。我添加的任务不会引发任何错误，也不会从主题中读取任何消息。Yarn UI 将任务显示为已接受。不确定我在这里做错了什么。

这是课程

            public class MyTask implements StreamTask {

                @Override
                public void process(IncomingMessageEnvelope incomingMessageEnvelope, MessageCollector messageCollector, TaskCoordinator taskCoordinator) throws Exception {

                    System.out.println(" key - " + incomingMessageEnvelope.getKey()  + " | message " + incomingMessageEnvelope.getMessage());

                }


            }

这是属性文件

            # Job
            job.factory.class=org.apache.samza.job.yarn.YarnJobFactory
            job.name=addresses

            # YARN
            yarn.package.path=file://${basedir}/target/${project.artifactId}-${pom.version}-dist.tar.gz

            # Task
            task.class=samza.examples.wikipedia.task.MyTask
            task.inputs=addressestopic

            # Serializers
            serializers.registry.json.class=org.apache.samza.serializers.JsonSerdeFactory

            # Kafka System
            systems.kafka.samza.factory=org.apache.samza.system.kafka.KafkaSystemFactory
            systems.kafka.samza.msg.serde=json
            systems.kafka.consumer.zookeeper.connect=localhost:2181/
            systems.kafka.producer.bootstrap.servers=localhost:9092

            # Job Coordinator
            job.coordinator.system=kafka
            # Add configuration to disable checkpointing for this job once it is available in the Coordinator Stream model
            # See https://issues.apache.org/jira/browse/SAMZA-465?focusedCommentId=14533346&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-14533346 for more details
            job.coordinator.replication.factor=1
4

1 回答 1

3

如果 Yarn UI 显示您的作业处于“接受”状态而不是“正在运行”，那么 Yarn 可能还没有找到资源来运行您的 Samza 作业。

通常，我注意到当您正在执行的本地机器上的磁盘空间不足时会发生这种情况。

  • 您可以检查 Yarn UI (localhost:8088) 并验证“活动节点”（在顶部）== 1 吗？
  • 此外，您可以交叉检查“丢失节点”的数量是否为零。如果有Lost Nodes，您可以点击链接查看为什么它无法使用。
于 2017-03-20T02:45:28.670 回答