KafkaでflatMapValuesを使用する方法

Question

ScalaでflatMapValuesをKafkaライブラリで使用するとエラーが発生します。ここに私のコードは次のとおりです。

val builder: KStreamBuilder = new KStreamBuilder() 
val textLines: KStream[String, String] = builder.stream("streams-plaintext-input") 
import collection.JavaConverters.asJavaIterableConverter 
val wordCounts: KTable[String, JLong] = textLines 
    .flatMapValues(textLine => textLine.toLowerCase.split("\\W+").toIterable.asJava) 
    .groupBy((_, word) => word) 
    .count("word-counts") 

と私はflatMapValues内textLineのエラーmissing parameter typeを取得しています。 flatMapValues((textLine: String) => textLine.toLowerCase.split("\\W+").toIterable.asJava)に置き換えても、それでも動作しません。

誰か知っていますか？私はこのような解決のScala 2.12.4での作業 おかげで、フェリペ

Answer 1

：私は推測

val props = new Properties 
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-wordcount") 
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092") 
    props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String.getClass.getName) 
    props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String.getClass.getName) 

    val stringSerde: Serde[String] = Serdes.String() 
    val longSerde: Serde[Long] = Serdes.Long() 

    val builder = new StreamsBuilder() 
    val textLines: KStream[String, String] = builder.stream("streams-plaintext-input") 

    val topology: Topology = builder.build() 

    println(topology.describe()) 

    val wordCounts: KTable[String, Long] = textLines 
    .flatMapValues { textLine => 
     println(textLine) 
     println(topology.describe()) 
     textLine.toLowerCase.split("\\W+").toIterable.asJava 
    } 
    .groupBy((_, word) => word) 
    // this is a stateful computation config to the topology 
    .count("word-counts") 

    wordCounts.to(stringSerde, longSerde, "streams-wordcount-output") 

    val streams = new KafkaStreams(topology, props) 
    streams.start()