flink+kafka实例

向右看齐 2022-10-12 12:23 200阅读 0赞

###  ###

*   *  1、kafka集群搭建
     *  2、java工程部署
     *  3、创建主类 FlinkReadKafka
     *  4、启动 zk
     *  5、启动 kafka 集群
     *  6、创建topic
     *  7、运行代码
     *  8、创建生产者和消费者

## 1、kafka集群搭建 ##

zookeeper集群以及kafka集群搭建详见博客：[kafka集群搭建][kafka]。只是本文将kafka版本升级为0.11.0.3。

## 2、java工程部署 ##

java maven 工程大部署详见[基于flink实现的worldcount][flink_worldcount]，本位只需要讲maven 工程pom配置文件增加 kafka 连接器  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3doYW5kZ2Ro_size_16_color_FFFFFF_t_70]

## 3、创建主类 FlinkReadKafka ##

代码如下：

public class FlinkReadKafka { 
        public static void main(String[] args) throws Exception { 
            StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
            Properties properties=new Properties();
            properties.setProperty("bootstrap.servers","node1:9092,node2:9092,node3:9092");
            properties.setProperty("group.id", "mykafkaxxx");
            FlinkKafkaConsumer011 kafka=new FlinkKafkaConsumer011("FlinkTopic",new SimpleStringSchema(),properties);
            DataStreamSource dss = env.addSource(kafka);
            SingleOutputStreamOperator flatMap = dss.flatMap(new FlatMapFunction<String, String>() { 
                @Override
                public void flatMap(String value, Collector<String> out) throws Exception { 
                    String[] split = value.split(" ");
                    for (String word : split) { 
                        out.collect(word);
                    }
                }
            });
            SingleOutputStreamOperator map = flatMap.map(new MapFunction<String, Tuple2<String, Integer>>() { 
                @Override
                public Tuple2<String, Integer> map(String word) throws Exception { 
                    return new Tuple2<>(word, 1);
                }
            });
            WindowedStream<Tuple2<String,Integer>,Tuple,TimeWindow> windowedStream = map.keyBy(0).timeWindow(Time.seconds(5));
            SingleOutputStreamOperator<Tuple2<String,Integer>>sum=windowedStream.sum(1);
            FlinkKafkaProducer011<String> sink = new FlinkKafkaProducer011<>("FlinkResult", new SimpleStringSchema(), properties);
            sum.map(new MapFunction<Tuple2<String,Integer>, String>() { 
    
                @Override
                public String map(Tuple2<String, Integer> tp2) throws Exception { 
                    return tp2.f0+"="+tp2.f1;
                }
            }).addSink(sink);
            env.execute();
    
        }
    }

## 4、启动 zk ##

在node1,node2,node3上

cd /opt/app/zookeeper-3.4.6/conf
     zkServer.sh  start

## 5、启动 kafka 集群 ##

在node1,node2,node3上

cd /opt/app/kafka_2.10-0.9.0.1/
     kafka-server-start.sh config/server.properties

## 6、创建topic ##

创建 source topic

kafka-topics.sh --create  --partitions 3  -replication-factor 2 --topic Flink_Topic --zookeeper node1:2181,node2:2181,node3:2181

创建 sink topic

kafka-topics.sh --create  --partitions 3  -replication-factor 2 --topic Flink_Result --zookeeper node1:2181,node2:2181,node3:2181

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3doYW5kZ2Ro_size_16_color_FFFFFF_t_70 1]

## 7、运行代码 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3doYW5kZ2Ro_size_16_color_FFFFFF_t_70 2]

## 8、创建生产者和消费者 ##

在node1 上创建生产者

kafka-console-producer.sh   --broker-list  node1:9092,node2:9092,node3:9092 --topic Flink_Topic

在node2 上创建 消费者

kafka-console-consumer.sh  --bootstrap-server node1:9092,node2:9092,node3:9092 --topic Flink_Result

生产者上输入字符串，可以看到 消费者上显示结果为经过 FlinkReadKafka处理后的结果  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3doYW5kZ2Ro_size_16_color_FFFFFF_t_70 3]

[kafka]: https://blog.csdn.net/whandgdh/article/details/116279215?spm=1001.2014.3001.5501
[flink_worldcount]: https://blog.csdn.net/whandgdh/article/details/117719543?spm=1001.2014.3001.5501
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3doYW5kZ2Ro_size_16_color_FFFFFF_t_70]: /images/20221005/7b26c7849c6f4f0988758f88a14d3928.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3doYW5kZ2Ro_size_16_color_FFFFFF_t_70 1]: /images/20221005/6d6796309a5b43f89efbcd8904f84c70.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3doYW5kZ2Ro_size_16_color_FFFFFF_t_70 2]: /images/20221005/d3a4746187d14854b9b1b62ce0b67f1d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3doYW5kZ2Ro_size_16_color_FFFFFF_t_70 3]: /images/20221005/9e5b19d18dda4ca99598277f271c2941.png