Flink讀取Kafka數據，進行彙總

今天介紹用 Flink 讀取Kafka生成的數據，並進行彙總的案例

第一步：環境準備，kafka，flink，zookeeper。我這邊是用的CDH環境，kafka跟zookeeper 都安裝完畢，並測試能夠正常使用

第二步：用kafka建立一個生產者進行消息生產

./kafka-console-producer.sh --broker-list 192.168.58.177:9092 --topic my_topic

　　

 

3.  第三步：在idea裏面建立一個flink項目。代碼以下：

　　　　 StreamExecutionEnvironment Env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.58.177:9092");
        properties.setProperty("zookeeper.connect", "192.168.58.171:2181,192.168.58.177:2181");
        properties.setProperty("group.id", "test");

        FlinkKafkaConsumer<String> myConsumer = new FlinkKafkaConsumer<String>("my_topic",new SimpleStringSchema(),properties);

        myConsumer.setStartFromLatest();
        myConsumer.setStartFromGroupOffsets();


        Env.setParallelism(2).setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<Tuple2<String,Integer>> stream = Env.addSource(myConsumer)
                .flatMap((String lines, Collector<Tuple2<String,Integer>> out) ->
                        Stream.of(lines.split(","))
                        .forEach(a -> out.collect(Tuple2.of(a,1))))
                .returns(Types.TUPLE(Types.STRING,Types.INT))
                .keyBy(0)
                //.window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sum(1)
                ;

        //stream.writeAsText("C:\\Users\\yaowentao\\Desktop\\a");
        stream.print();
        Env.execute("my first stream flink");

　　

第四步：返回kafka進行消息輸入，並觀察控制檯是否有數據輸出

 

 

 

 

這樣就能初步實現 flink讀取kafka的消息