Kafka中steamAPi操作
生活随笔
收集整理的這篇文章主要介紹了
Kafka中steamAPi操作
小編覺得挺不錯的,現在分享給大家,幫大家做個參考.
/*
用于實現將一個topic的數據讀取后從轉換成大寫,再將結果寫入另一個topic*/
public class StreamAPI {public static void main(String[] args) {Properties props = new Properties();//設置程序的唯一標識props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-application");//設置kafka集群props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "node01:9092,node02:9092,node03:9092");//設置序列化與反序列化props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());//實例一個計算邏輯StreamsBuilder streamsBuilder = new StreamsBuilder();//設置計算邏輯 stream:讀取 to:寫入streamsBuilder.stream("18BD34").mapValues(line->line.toString().toUpperCase()).to("18BD34-1");//構建Topology對象(拓撲,流程)final Topology topology = streamsBuilder.build();//實例 kafka流KafkaStreams streams = new KafkaStreams(topology, props);//啟動流計算streams.start();}}
總結
以上是生活随笔為你收集整理的Kafka中steamAPi操作的全部內容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: linux构建widi显示服务,如此简单
- 下一篇: 关于Excel操作编写的一个软件设计构思