Repository navigation
Expand file tree
/
Copy pathWordCountDSL.java
More file actions
63 lines (54 loc) · 2.75 KB
/
Copy pathWordCountDSL.java
File metadata and controls
63 lines (54 loc) · 2.75 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
package com.jasongj.kafka.stream;
import java.util.Arrays;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KStreamBuilder;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.Windowed;
public class WordCountDSL {
public static void main(String[] args) throws InterruptedException {
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-wordcount-dsl");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka0:19092");
props.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, "zookeeper0:12181/kafka");
props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
KStreamBuilder builder = new KStreamBuilder();
KStream<String, String> stream = builder.stream("words");
// stream.flatMapValues(values -> Arrays.asList(values.toLowerCase().split(" ")))
// .map((k, v) -> KeyValue.<String, String>pair(v, v)).groupByKey().aggregate(
// () -> 0L,
// (aggKey, value, aggregate) -> aggregate + 1L,
// TimeWindows.of(5000).advanceBy(1000),
// Serdes.Long(),
// "Counts1")
// .foreach((Windowed<String> window, Long value) -> {
// System.out.printf("key=%s, value=%s, start=%d, end=%d\n",window.key(), value, window.window().start(), window.window().end());
// });
KStream<String, String> kStream = stream.flatMapValues(values -> Arrays.asList(values.toLowerCase().split(" ")))
.map((k, v) -> KeyValue.<String, String>pair(v, v)).groupByKey().aggregate(
() -> 0L,
(aggKey, value, aggregate) -> aggregate + 1L,
TimeWindows.of(5000).advanceBy(5000),
Serdes.Long(),
"Counts")
.toStream()
.map((Windowed<String> window, Long value) -> {
return new KeyValue<String, String>(window.key(), String.format("key=%s, value=%s, start=%d, end=%d\n",window.key(), value, window.window().start(), window.window().end()));
});
kStream.to(Serdes.String(), Serdes.String(), "count");
// KTable<String, Long> kTable = stream.flatMapValues(values -> Arrays.asList(values.toLowerCase().split(" ")))
// .map((k, v) -> KeyValue.<String, String>pair(v, v)).groupByKey().count("Counts");
// kTable.to(Serdes.String(), Serdes.Long(), "stream-dsl-sink");
KafkaStreams streams = new KafkaStreams(builder, props);
streams.start();
Thread.sleep(100000L);
streams.close();
}
}