如何根据某些条件将字符串拆分为不同的kafka主题

Question

我正在尝试根据条件将字符串拆分为不同的 kafka 主题。
这是拓扑。

1. 将字符串拆分为单词。
2. 根据条件匹配每个单词（这里是一组好词和一组坏词）
3. 如果在字符串中找到至少 1 个来自 Bad words set 的单词，它将被发送到 Bad-string
   主题，否则将被发送到好字符串主题。

问题：

每个字符串都只针对一个主题。 （坏字符串主题）

输入：

1. 你们的服务很好。

2. 他又气又伤心。

3. 您的服务很糟糕，但我仍然很高兴。

输出：
好字符串（主题）

你们的服务很好。 （它包含好词，“好”）

错误字符串（主题）

1. 他又生气又伤心（里面有不好的词，“生气”和“伤心”）
2. 您的服务很糟糕，但我仍然很高兴。 （虽然，有一个 好词“快乐”，但至少有一个坏词“坏”）

代码如下：

@Configuration
@Slf4j
public class SplitSentence {
    
    private static final Set<String> BAD_WORDS = Set.of("angry", "sad", "bad");
    private static final Set<String> GOOD_WORDS = Set.of("happy", "good", "helpful");

    @SuppressWarnings("unchecked")
    @Bean
    public KStream<String,String> windowCount(StreamsBuilder builder)
    {
        var stringSerde = Serdes.String();
        var jsonSerde = new JsonSerde<>(CountingDemo.class);

        ((JsonDeserializer) jsonSerde.deserializer()).setUseTypeHeaders(false);

        var input = builder.stream("counting",Consumed.with(stringSerde,jsonSerde));

        var feedbackStreams = input.flatMap(splitWords()).branch(isGoodWord(), isBadWord());
        
        boolean newString = feedbackStreams[1].toString().isEmpty();

        if(newString)
            input.to("good-string");
        else    
            input.to("bad-string");

        return input;
    }

    private Predicate<? super String, ? super String> isBadWord() {
        return (key, value) -> BAD_WORDS.contains(value);
    }

    private Predicate<? super String, ? super String> isGoodWord() {
        return (key, value) -> GOOD_WORDS.contains(value);
    }

    private KeyValueMapper<String, CountingDemo, Iterable<KeyValue<String,String>>> splitWords() 
    {
        return (key,value) -> Arrays
                            .asList(value.getText().replaceAll("[^a-zA-Z ]", "").toLowerCase().split("\\s+")).stream()
                            .distinct().map(word -> KeyValue.pair(value.getText(), word)).collect(Collectors.toList());
    }
}

   
     

CountingDemo.java

public class CountingDemo {
    
    private String name;
    private String text;
}    
  

我哪里错了？
有没有更好的逻辑？

Answer 1

if 语句始终为 false，因为 KStream 对象的 .toString() 是它的元数据，并且永远不会为空。

如果您想在两个主题之间拆分完整的原始字符串，则根本不应该使用平面映射。

话虽如此，看来你想要

var feedbackStreams = input.branch(hasGoodWords(), hasBadWords());

feedbackStreams[0].to("good-string");
feedbackStreams[1].to("bad-string");

这两个函数获取完整的输入消息并与集合进行比较，而不是给出单个单词。
虽然，我认为您只需要 一个 函数来捕获 所有带有好词的消息到good-string，以及所有其他消息（不好/bad，好/坏，还有一些坏）到bad-string 主题

例如

    var feedbackStreams = input.branch(this::hasOnlyGoodWords, (k,v) -> true);
    feedbackStreams[0].to("good-string");
    feedbackStreams[1].to("bad-string");
    return input;
}

private boolean hasOnlyGoodWords(Object key, String value) {
    String cleaned = value.getText().replaceAll("[^a-zA-Z ]", "").toLowerCase();
    Set<String> uniqueWords = Arrays.stream(cleaned.split("\\s+")).distinct().collect(toSet());
    for (String s : BAD_WORDS) {
        if (uniqueWords.contains(s)) return false;
    }
    uniqueWords.retainAll(GOOD_WORDS);
    return uniqueWords.size() > 0;
}