使用 kafka 流根据消息密钥向主题发送消息

Question

我希望能够根据消息键的键将 Kafkastream 中的所有记录发送到不同的主题。 前任。 Kafka 中的流包含名称作为键和记录作为值。我想根据记录的键将这些记录扇出到不同的主题

数据: (jhon -> {jhonsRecord}),(sean -> {seansRecord}),(mary -> {marysRecord}),(jhon -> {jhonsRecord2}), 预计

• topic1 :name: jhon ->(jhon -> {jhonsRecord}),(jhon -> {jhonsRecord2})
• topic2 :sean-> (sean -> {seansRecord})
• topic3 :mary -> (mary -> {marysRecord})

以下是我现在执行此操作的方式，但是由于名称列表很长，所以速度很慢。另外，即使记录了几个名字，我也需要遍历整个列表请提出修复建议

    for( String name : names )
    {
        recordsByName.filterNot(( k, v ) -> k.equalsIgnoreCase(name)).to(name);
    } 

Answer 1

我认为你应该使用KStream::to(final TopicNameExtractor<K, V> topicExtractor) 函数。它使您能够计算每条消息的主题名称。

示例代码：

final KStream<String, String> stream = ???;
stream.to((key, value, recordContext) -> key);

Answer 2

我认为您正在寻找的是KStream#branch。

以下内容未经测试，但显示了大致思路

// get a list of predicates to branch a topic on
final List<String> names = Arrays.asList("jhon", "sean", "mary");
final Predicate[] predicates = names.stream()
    .map((Function<String, Predicate<String, Object>>) n -> (s, o) -> s.equals(n))
    .toArray(Predicate[]::new);

// example input
final KStream<Object, Object> stream = new StreamsBuilder().stream("names");

// split the topic
KStream<String, Object>[] branches = stream.branch(predicates);
for (int i = 0; i < names.size(); i++) {
    branches[i].to(names.get(i));
}

// KStream branches[0] contains all records whose keys are "jhon"
// KStream branches[1] contains all records whose keys are "sean"
...

Answer 3

如果您需要为每个用户生成聚合数据，则无需为每个用户写入单独的主题。你最好在源流上写一个聚合。这样一来，您不会最终每个键都有一个主题，但您仍然可以独立地对每个用户运行操作。

Serde<UserRecord> recordSerde = ...
KStream<Stream, UserAggregate> aggregateByName = recordsByName
   .groupByKey(Grouped.with(Serdes.String(), recordSerde))
   .aggregate(...)
   .toStream()

详情请见https://docs.confluent.io/current/streams/developer-guide/dsl-api.html#aggregating

这种方法将扩展到数百万用户，您目前无法通过每个用户一个主题的方法来实现这一目标。