个人中心
文章发布
退出
作者热门文章
- Java 双重比较
- java - 比较器与 Apache BeanComparator
- Objective-C 完成 block 导致额外的方法调用？
- database - RESTful URI 是否应该公开数据库主键？
24
4
我们有一个从 Gnip API 读取并将推文发送到 Kafka 集群的 Spark 流作业。
Kafka 集群是使用 Cloudera Manager 安装的。
有时，cloudera 管理器会显示某些 Kafka 节点的健康状况不佳消息。健康状况不佳消息与 NTP 服务有关。某些节点突然与 NTP 服务器不同步。
一旦发生这种情况，Spark streaming 作业就会卡住，大量作业排队等待很长时间。
为什么Kafka节点与NTP服务器的同步会影响spark streaming作业中的Kafka producer？
最佳答案
每个分区在 Kafka 代理中都有其领导者和追随者，Kafka 通过它们提供容错能力。该机制基于使用NTP服务的ZooKeeper。
如果你使用默认配置，leader会收到你的数据，并尽量写入followers。在数据写入每个跟随者之前，它不会响应成功消息。所以您的 Spark 应用程序会阻塞。
您还可以更改您的 Kafka 配置以在领导者收到数据时立即响应或在领导者将数据写入磁盘时立即响应。
您可以在 Kafka documents 中找到更多信息.
关于java - 如果某些 Kafka 节点时间偏移不同步，Spark 流作业会卡住，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/39312317/
24
4
0
我在 Windows 机器上启动 Kafka-Server 时出现以下错误。我已经从以下链接下载了 Scala 2.11 - kafka_2.11-2.1.0.tgz:https://kafka.ap
关于Apache-Kafka messaging queue . 我已经从 Kafka 下载页面下载了 Apache Kafka。我已将其提取到 /opt/apache/installed/kafka
假设我有 Kafka 主题 cars。 我还有一个消费者组 cars-consumers 订阅了 cars 主题。 cars-consumers 消费者组当前位于偏移量 89。 当我现在删除 cars
我想知道什么最适合我:Kafka 流或 Kafka 消费者 api 或 Kafka 连接？ 我想从主题中读取数据，然后进行一些处理并写入数据库。所以我编写了消费者，但我觉得我可以编写 Kafka 流应
我曾研究过一些 Kafka 流应用程序和 Kafka 消费者应用程序。最后，Kafka流不过是消费来自Kafka的实时事件的消费者。因此，我无法弄清楚何时使用 Kafka 流或为什么我们应该使用
Kafka Acknowledgement 和 Kafka 消费者 commitSync() 有什么区别 两者都用于手动偏移管理，并希望两者同步工作。 请协助 最佳答案 使用 spring-kafka
如何在 Kafka 代理上代理 Apache Kafka 生产者请求，并重定向到单独的 Kafka 集群？ 在我的特定情况下，无法更新写入此集群的客户端。这意味着，执行以下操作是不可行的: 更新客户端
我需要在 Kafka 10 中命名我的消费者，就像我在 Kafka 8 中所做的一样，因为我有脚本可以嗅出并进一步使用这些信息。 显然，consumer.id 的默认命名已更改(并且现在还单独显示了
1.概述 我们会看到zk的数据中有一个节点/log_dir_event_notification/，这是一个序列号持久节点 这个节点在kafka中承担的作用是: 当某个Broker上的LogDir出现
我正在使用以下命令: bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test.topic --property
我很难理解 Java Spring Boot 中的一些 Kafka 概念。我想针对在服务器上运行的真实 Kafka 代理测试消费者，该服务器有一些生产者已将数据写入/已经将数据写入各种主题。我想与服务
我的场景是我使用了很多共享前缀的 Kafka 主题(例如 house.door, house.room ) 并使用 Kafka 流正则表达式主题模式 API 使用所有主题。 一切看起来都不错，我得到了
有没有办法以编程方式获取kafka集群的版本？例如，使用AdminClient应用程序接口(interface)。 我想在消费者/生产者应用程序中识别 kafka 集群的版本。 最佳答案 目前无法检索
每当我尝试重新启动 kafka 时，它都会出现以下错误。一旦我删除/tmp/kafka-logs 它就会得到解决，但它也会删除我的主题。 有办法解决吗？ ERROR Error while
我是 Apache Kafka 的新用户，我仍在了解内部结构。 在我的用例中，我需要从 Kafka Producer 客户端动态增加主题的分区数。 我发现了其他类似的 questions关于增加分区大
正如 Kafka 文档所示，一种方法是通过 kafka.tools.MirrorMaker 来实现这一点。但是，我需要将一个主题(比如 测试 带 1 个分区)(其内容和元数据)从生产环境复制到没有连接
我已经在集群中配置了 3 个 kafka，我正在尝试与 spring-kafka 一起使用。 但是在我杀死 kafka 领导者之后，我无法将其他消息发送到队列中。 我将 spring.kafka.bo
我的 kafka sink 连接器从多个主题(配置了 10 个任务)读取，并处理来自所有主题的 300 条记录。根据每个记录中保存的信息，连接器可以执行某些操作。 以下是触发器记录中键值对的示例: "
我有以下 kafka 流代码 public class KafkaStreamHandler implements Processor{ private ProcessorConte
当 kafka-streams 应用程序正在运行并且 Kafka 突然关闭时，应用程序进入“等待”模式，发送警告日志的消费者和生产者线程无法连接，当 Kafka 回来时，一切都应该(理论上)去恢复正常
我是一名优秀的程序员,十分优秀！