我在某个主题中有多条消息（更确切地说是日志消息），这些消息对一个消息块具有相同的id（这些id不断变化，但对某个消息块保持不变），我需要找到一种方法，将所有具有该id的消息分组，或者在所有主题之间共享包含在具有相同id的消息中的数据消费者群体中的消费者。那么，我有没有办法在一个消费群体中的不同消费者之间共享数据呢？
z31licg01#
这听起来像是一个会话用例。kafka没有提供任何将消息分组或嵌套在一起的方法，因此您必须自己这样做：在处理消息组时将状态保存在consumer中，并使用某种标头 Package 消息组。然后您可以将其推到 Package 消息组的新主题。更好的办法可能是利用外部数据库或其他系统，采用更灵活的方法根据字段选择或组织数据。你可以看看这篇博客文章中使用spark streaming+hbase的例子。
68de4m5k2#
有两种方法可以做到这一点。当您发布消息本身时，使用分区键创建一个消息，这样所有具有相同id的消息都将进入单个分区。那么在消费者方面，它将始终由单个消费者消费[https://cwiki.apache.org/confluence/display/kafka/0.8.0+producer+example]如果在用户端使用spark流，则可以使用滑动窗口概念对所有相同的id消息进行分组[http://spark.apache.org/docs/latest/streaming-programming-guide.html#window-[操作]
2条答案
按热度按时间z31licg01#
这听起来像是一个会话用例。kafka没有提供任何将消息分组或嵌套在一起的方法，因此您必须自己这样做：在处理消息组时将状态保存在consumer中，并使用某种标头 Package 消息组。然后您可以将其推到 Package 消息组的新主题。
更好的办法可能是利用外部数据库或其他系统，采用更灵活的方法根据字段选择或组织数据。你可以看看这篇博客文章中使用spark streaming+hbase的例子。
68de4m5k2#
有两种方法可以做到这一点。
当您发布消息本身时，使用分区键创建一个消息，这样所有具有相同id的消息都将进入单个分区。那么在消费者方面，它将始终由单个消费者消费[https://cwiki.apache.org/confluence/display/kafka/0.8.0+producer+example]
如果在用户端使用spark流，则可以使用滑动窗口概念对所有相同的id消息进行分组[http://spark.apache.org/docs/latest/streaming-programming-guide.html#window-[操作]