问题标签 [apache-kafka]

我正在使用 Kafka 0.8 beta，我只是想弄乱发送不同的对象，使用我自己的编码器对它们进行序列化，然后将它们发送到现有的代理配置。现在我正试图让 DefaultEncoder 正常工作。

我有代理和所有设置并为 StringEncoder 工作，但我无法获得任何其他数据类型，包括纯字节 []，以由代理发送和接收。

我的生产者代码是：

我使用了与此处给出的示例相同的 SimplePartitioner ，并用字符串替换所有字节数组并将序列化程序更改为 kafka.serializer.StringEncoder 工作得很好。

供参考，SimplePartitioner：

我究竟做错了什么？

Given the fact that a Consumer in Kafka can be configured to receive messages only from one partition, I haven't seen a configuration approach for this.

How do I tell a Kafka broker:

That I'm a CONSUMER A with this ID, under this GROUP L, subscribing to TOPIC X, willing to get the stream of PARTITION A?

And then, start another CONSUMER B, under GROUP L, which subscribes to TOPIC X, willing to get the stream of PARTITION B?

In other words, as in the scenario described by Kafka,

How do I start a consumer subscribing to a wall feed of a specific user? Can I start a consumer when a user logs in, consumer his/her partition form the topic and send the feed to the client?

我有一个 3 节点 kafka 集群设置。我正在使用storm来阅读来自kafka的消息。我系统中的每个主题都有 7 个分区。

现在我面临一个奇怪的问题。直到 3 天前，一切正常。但是，现在看来我的风暴拓扑无法专门从 2 个分区 - #1 和 #4 读取。

我试图深入研究这个问题，发现在我的 kafka 日志中，对于这两个分区，缺少一个偏移量，即在 5964511 之后，下一个偏移量是 5964513 而不是 5964512。

由于缺少偏移量，简单消费者无法继续进行下一个偏移量。我做错了什么还是一个已知的错误？

这种行为可能是什么原因？

我正在使用以下代码来读取有效偏移量的窗口：

这给了我以下输出：

因此，我提供的偏移量正好在偏移范围内。

我正在寻找可用于将数据从社交媒体流式传输到 hadoop 的技术。我搜索并找到了那些技术

1. 水槽。
2. 风暴。
3. 卡夫卡。

哪个工具最好？为什么？有人熟悉其他工具吗？

用 kafka（java 代码）做我的第一步我想为 kafka 生产者创建一个简单的测试，像这样我可以模拟 zoo keeper（这个实现看起来不错，但我无法访问那里的一些类，特别是 EmbeddedZookeeper和TestUtils）。

有任何想法吗？

在由 maven 构建的 Java 项目中，我正在尝试创建一个基本的 kafka（生产者消费者）示例。运行此代码时，我收到此异常：

我还尝试对真正的 kafka 服务器（使用 zoo-keeper）运行类似的代码，然后再次 - 消息已成功发送到 kafka（我也可以在我的 kafka 服务器中验证它）但是当尝试使用它时，它得到了同样的异常。

因为我怀疑这是 jar 版本不匹配的问题，所以这里是我的 pom.xml：

有任何想法吗？

我刚刚创建了一个新主题，还没有产生任何消息。00000000000000000000.index在目录中创建了一个名为 in 的文件/tmp/kafka-logs-1/topicname-0/，该文件的大小非常大。我用vi打开那个二进制文件，内容只有“0000 0000 0000 0000...”这是什么意思？这个索引文件是关于什么的？

我想设计一个具有向 Kafka 发送消息的应用程序浏览器的系统。消费者应该以 avro 格式将事件写入 HDFS。

这应该是什么样子？

什么时候是确认消息的好时机？

我如何才能附加到 hdfs 文件？我阅读了有关 cdk 的信息，它看起来像是朝着好的方向发展。我不能是如何/何时刷新记录。

有没有人有一个很好的例子？

我正在研究 kafka 以实现低延迟消息队列，并且我一直在阅读有关消费者长轮询的信息。但是，没有关于如何实际使用长轮询或需要设置哪些选项来启用它的示例。如何使用 kafka java api 启用长轮询？

我正在尝试使用 gevent 在 python 2.7 上使用 brod 写入 kafka 0.7.2。

这是我收到的错误消息。猜猜是因为阻塞。brod 支持龙卷风，但我使用 gevent。

我尝试使用 gevent-kakfa 但依赖于 gevent-zookeeper。

尝试连接到 zookeeper 时，我收到以下消息：

没有我可以使用有效的 gevent 编写消息的 python 库吗？