Kafka---窗口函数
Posted Shall潇
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Kafka---窗口函数相关的知识,希望对你有一定的参考价值。
- 跳跃窗口
- 滑动窗口
- 会话窗口
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
/**
* @Author shall潇
* @Date 2021/5/28
* @Description
*/
public class WindowStreamDemo {
public static void main(String[] args) {
Properties properties = new Properties();
properties.put(StreamsConfig.APPLICATION_ID_CONFIG,"windowdemo");
properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.159.100:9092");
properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
properties.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG,3000);
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);
StreamsBuilder builder = new StreamsBuilder();
// TimeWindowedKStream<String, String> windowsdemo = builder.stream("windowsdemo")
// .flatMapValues((value) -> {
// String[] split = value.toString().split("\\\\s");
// return Arrays.asList(split);
// }).map((key, value) -> {
// return new KeyValue<String, String>(value, "1");
// }).groupByKey()
//第一种 跳跃窗口 ,每次移动15秒
// .windowedBy(TimeWindows.of(Duration.ofSeconds(15).toMillis()));
//第二种 滑动窗口,窗口大小为 15秒,每次移动5秒
// .windowedBy(TimeWindows.of(Duration.ofSeconds(15).toMillis()).advanceBy(Duration.ofSeconds(5).toMillis()));
//第三种 会话窗口,固定时间间隔sessionWindow 返回值 SessionWindowedKStream
SessionWindowedKStream<String, String> windowsdemo = builder.stream("windowsdemo")
.flatMapValues((value) -> {
String[] split = value.toString().split("\\\\s");
return Arrays.asList(split);
}).map((key, value) -> {
return new KeyValue<String, String>(value, "1");
}).groupByKey()
.windowedBy(SessionWindows.with(Duration.ofSeconds(15).toMillis()));
KStream<Windowed<String>, Long> wlks = windowsdemo.count().toStream();
wlks.foreach((key,value)->{
System.out.println(key+"-->"+value);
});
Topology topo = builder.build();
//将 拓扑结构 和 配置信息 放入 KafkaStream
KafkaStreams kafkaStreams = new KafkaStreams(topo, properties);
CountDownLatch latch = new CountDownLatch(1);
Runtime.getRuntime().addShutdownHook(new Thread("线程1"){
@Override
public void run() {
kafkaStreams.close();
latch.countDown();
}
});
kafkaStreams.start();
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.exit(0);
}
}
以上是关于Kafka---窗口函数的主要内容,如果未能解决你的问题,请参考以下文章