Flink：将Kafka中的数据作为Flink的数据源

系统管理员 2024-03-23 12:57 78阅读 0赞

### 启动kafka ###

./kafka-server-start.sh -daemon ../config/server.properties

### 创建topic ###

bin/kafka-topics.sh --create --zookeepper 192.168.25.129:2181 --replication-factor 1 --partitions 1 --topic test

### 启动控制台kafka生产者 ###

bin/kafka-console-producer.sh --broker-list 192.168.25.129:9092 --topic test

### 依赖 ###

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>1.7.2</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.12</artifactId>
        <version>1.7.2</version>
        <scope>provided</scope>
    </dependency>	
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka_2.11</artifactId>
        <version>1.7.2</version>
    </dependency>

### 代码 ###

import org.apache.flink.api.common.functions.FlatMapFunction;
    import org.apache.flink.api.common.serialization.SimpleStringSchema;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
    import org.apache.flink.util.Collector;
     
    import java.util.Properties;
     
    public class StreamFromKafka {
        public static void main(String[] args) throws Exception {
     
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            Properties properties = new Properties();
     
            properties.setProperty("bootstrap.servers","192.168.25.129:9092");
            // 参数一：主题名
            // 参数二：反序列化约束，以便于Flink决定如何反序列化从Kafka获得的数据
            // 参数三：消费者的一些配置信息
            FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<String>("test", new SimpleStringSchema(), properties);
            
            DataStreamSource<String> data = env.addSource(consumer);
            SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = data.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                    for (String word : s.split(" ")) {
                        collector.collect(Tuple2.of(word, 1));
                    }
                }
            });
            
            SingleOutputStreamOperator<Tuple2<String, Integer>> result = wordAndOne.keyBy(0).sum(1);
            result.print();
            env.execute();
        }
    }

**Flink：将Flink中的数据下沉到Kafka**

> [https://blog.csdn.net/weixin\_45427648/article/details/130181548][https_blog.csdn.net_weixin_45427648_article_details_130181548]

[https_blog.csdn.net_weixin_45427648_article_details_130181548]: https://blog.csdn.net/weixin_45427648/article/details/130181548