一、基于前面kafka部署

大数据-玩转数据-Kafka安装

二、FLINK中编写代码

package com.lyh.flink04;import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;import java.util.Properties;public class flink04_fromkafka {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);Properties properties = new Properties();properties.setProperty("bootstrap.servers","hadoop100:9092");properties.setProperty("group.id","test");env.addSource(new FlinkKafkaConsumer<>("wordsendertest",new SimpleStringSchema(),properties)).print();env.execute();}
}

三、运行测试

运行本段代码，等待kafka产生数据进行消费。