Hi Matthias,
Thank you for the quick response, appreciate it!
I created the topics wordCount-input and wordCount-output. Pushed some data
to wordCount-input using
docker exec -it $(docker ps -f "name=kafka\\." --format "{{.Names}}")
/bin/kafka-console-producer --broker-list localhost:9092 --topic
wordCount-input
test
new
word
count
wordcount
word count
So, when I check the number of messages in wordCount-input I see the same
messages. However, when I run below code I do not see any message/data in
wordCount-output.
Can I connect to kafka in VM/docker container using below code or do I need
to change/add other parameters? How can I submit the code to
kafka/kafka-connect? Do we have similar concept as SPARK to submit the
code(e.g. jar file)?
I really appreciate your input as I am blocked and cannot run even below
simple example.
Best regards,
Mina
I changed the code to be as below:
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordCount-streaming");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "<ipAddress>:9092");
props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG,
Serdes.String().getClass().getName());
props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG,
Serdes.String().getClass().getName());
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10);
props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
// setting offset reset to earliest so that we can re-run the demo
code with the same pre-loaded data
// Note: To re-run the demo, you need to use the offset reset tool:
//
https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Streams+Application+Reset+Tool
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
KStreamBuilder builder = new KStreamBuilder();
KStream<String, String> source = builder.stream("wordCount-input");
KTable<String, Long> counts = source
.flatMapValues(new ValueMapper<String, Iterable<String>>() {
@Override
public Iterable<String> apply(String value) {
return
Arrays.asList(value.toLowerCase(Locale.getDefault()).split(" "));
}
}).map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
@Override
public KeyValue<String, String> apply(String key, String value) {
return new KeyValue<>(value, value);
}
})
.groupByKey()
.count("Counts");
// need to override value serde to Long type
counts.to(Serdes.String(), Serdes.Long(), "wordCount-output");
KafkaStreams streams = new KafkaStreams(builder, props);
streams.start();
// usually the stream application would be running forever,
// in this example we just let it run for some time and stop since the
input data is finite.
Thread.sleep(5000L);
streams.close();
On Mon, Mar 13, 2017 at 4:09 PM, Matthias J. Sax <matth...@confluent.io>
wrote:
> Maybe you need to reset your application using the reset tool:
> http://docs.confluent.io/current/streams/developer-
> guide.html#application-reset-tool
>
> Also keep in mind, that KTables buffer internally, and thus, you might
> only see data on commit.
>
> Try to reduce commit interval or disable caching by setting
> "cache.max.bytes.buffering" to zero in your StreamsConfig.
>
>
> -Matthias
>
> On 3/13/17 12:29 PM, Mina Aslani wrote:
> > Hi,
> >
> > This is the first time that am using Kafka Stream. I would like to read
> > from input topic and write to output topic. However, I do not see the
> word
> > count when I try to run below example. Looks like that it does not
> connect
> > to Kafka. I do not see any error though. I tried my localhost kafka as
> well
> > as the container in a VM, same situation.
> >
> > There are over 200 message in the input kafka topic.
> >
> > Your input is appreciated!
> >
> > Best regards,
> > Mina
> >
> > import org.apache.kafka.common.serialization.*;
> > import org.apache.kafka.streams.*;
> > import org.apache.kafka.streams.kstream.*;
> >
> > import java.util.*;
> > import java.util.regex.*;
> >
> > public class WordCountExample {
> >
> >
> > public static void main(String [] args) {
> > final Properties streamsConfiguration = new Properties();
> > streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG,
> > "wordcount-streaming");
> > streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
> > "<IPADDRESS>:9092");
> > streamsConfiguration.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG,
> > Serdes.String().getClass().getName());
> > streamsConfiguration.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG,
> > Serdes.String().getClass().getName());
> > streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG,
> > 10 * 1000);
> >
> > final Serde<String> stringSerde = Serdes.String();
> > final Serde<Long> longSerde = Serdes.Long();
> >
> > final KStreamBuilder builder = new KStreamBuilder();
> >
> > final KStream<String, String> textLines =
> > builder.stream(stringSerde, stringSerde, "wordcount-input");
> >
> > final Pattern pattern = Pattern.compile("\\W+",
> > Pattern.UNICODE_CHARACTER_CLASS);
> >
> > final KStream<String, Long> wordCounts = textLines
> > .flatMapValues(value ->
> > Arrays.asList(pattern.split(value.toLowerCase())))
> > .groupBy((key, word) -> word)
> > .count("Counts")
> > .toStream();
> >
> >
> > wordCounts.to(stringSerde, longSerde, "wordcount-output");
> >
> > final KafkaStreams streams = new KafkaStreams(builder,
> > streamsConfiguration);
> > streams.cleanUp();
> > streams.start();
> >
> > Runtime.getRuntime().addShutdownHook(new
> Thread(streams::close)); }
> > }
> >
>
>
