KStream 不兼容的类型

Question

我正在尝试使用字符串在编码部分进行一些操作，并提出了以下代码，请帮助解决代码的语法问题。

KStreamBuilder builder = new KStreamBuilder();
KStream<String, String> textlines = builder.stream("INTOPIC");
KStream<String, String> mstream = textlines
    .foreach(new ForeachAction<String, String>(){
        public void apply(String key, String value){
            String str=value.replace("[","");
            for(int y=0; y<str.length(); y++){
                //System.out.print(str.charAt(y));
            }
            //System.out.print("\n");
        }
    });
mstream.to("OUTTOPIC");

在第二个 KStream 语句中显示不兼容的类型..... 我希望替换逻辑可以使用，它应该会影响 textlines 变量。提前致谢。

Answer 1

foreach 的结果是 void，而不是下一个 KStream。如果要转换数据，则需要使用 map 代替...

代码看起来如下（特别是如果您还需要更改密钥，如果不需要 - 请参阅下一个变体）：

KStreamBuilder builder = new KStreamBuilder();
KStream<String, String> textlines = builder.stream("INTOPIC");
KStream<String, String> mstream = textlines
    .map((key, value) -> new KeyValue<>(key, value.replace("[",""))));
mstream.to("OUTTOPIC");

或者您可以将第 3-4 行替换为以下内容（这将导致更好的执行）：

KStream<String, String> mstream = textlines
    .mapValues(value -> value.replace("[",""));

KStream 不兼容的类型

KStream incompatible types

apache-kafka

apache-kafka-streams