4.1.10 Flink-流处理框架-Flink流处理API之Source数据源

我不是女神ヾ 2022-10-31 01:41 259阅读 0赞

**目录**

1.写在前面

2.Source数据源

2.1 从集合中读取数据

2.2 从文件中读取数据

2.3 以kafka消息队列的数据作为数据来源

2.4 自定义Source

--------------------

# 1.写在前面 #

我们在创建完成环境之后，我们需要读取数据，我们可以从哪些地方读取数据呢？我们区分为集合、文件、kafka消息队列、自定义数据源。

# 2.Source数据源 #

## 2.1 从集合中读取数据 ##

首先创建一个javaBean，传感器温度读数的数据类型**SensorReading**

package com.atguigu.apitest.beans;
    // 传感器温度读数的数据类型
    public class SensorReading {
        // 属性：id，时间戳，温度值
        private String id;
        private Long timestamp;
        private Double temperature;
    
        public SensorReading() {
        }
    
        public SensorReading(String id, Long timestamp, Double temperature) {
            this.id = id;
            this.timestamp = timestamp;
            this.temperature = temperature;
        }
    
        public String getId() {
            return id;
        }
    
        public void setId(String id) {
            this.id = id;
        }
    
        public Long getTimestamp() {
            return timestamp;
        }
    
        public void setTimestamp(Long timestamp) {
            this.timestamp = timestamp;
        }
    
        public Double getTemperature() {
            return temperature;
        }
    
        public void setTemperature(Double temperature) {
            this.temperature = temperature;
        }
    
        @Override
        public String toString() {
            return "SensorReading{" +
                    "id='" + id + '\'' +
                    ", timestamp=" + timestamp +
                    ", temperature=" + temperature +
                    '}';
        }
    }

从集合中读取数据相当于是把数据都写到代码中了，显然并不是一个非常好的方法。

执行结果如下所示：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70][]

## 2.2 从文件中读取数据 ##

package com.atguigu.apitest.source;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    public class SourceTest2_File {
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            // 从文件读取数据
            DataStream<String> dataStream = env.readTextFile("D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt");
    
            // 打印输出
            dataStream.print();
    
            env.execute();
        }
    }

执行结果是：

![2021022023281646.png][]

## 2.3 以kafka消息队列的数据作为数据来源 ##

因为flink数据流特点，我们知道数据源不会直接放在代码集合中，也不会放在一个固定的文件里面供我们读取，都是实时不断产生新数据。**我们一般利用flume等工具进行采集数据，将数据sink到kafka消息队列中，然后用flink读取kafka中数据进行消费，实时进行数据计算**。

kafka作为数据源，需要引入kafka连接器的依赖：pom.xml

<dependency>
     <groupId>org.apache.flink</groupId>
     <artifactId>flink-connector-kafka-0.11_2.12</artifactId>
     <version>1.10.1</version>
    </dependency>

具体代码如下：

package com.atguigu.apitest.source;=
    
    import org.apache.flink.api.common.serialization.SimpleStringSchema;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
    
    import java.util.Properties;
    
    public class SourceTest3_Kafka {
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            //kafka配置
            Properties properties = new Properties();
            properties.setProperty("bootstrap.servers", "localhost:9092");
            properties.setProperty("group.id", "consumer-group");
            properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            properties.setProperty("auto.offset.reset", "latest");
    
            // 从kafka读取数据
            DataStream<String> dataStream = env.addSource( new FlinkKafkaConsumer011<String>("sensor", new SimpleStringSchema(), properties));
    
            // 打印输出
            dataStream.print();
    
            env.execute();
        }
    }

启动kafka，并且创建kafka生产者进程，查看flink实时运行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1][]

## 2.4 自定义Source ##

除了以上的 source 数据来源，我们还可以自定义 source。需要做的，只是传入 一个 SourceFunction 就可以。具体调用如下：

DataStream<SensorReading> dataStream = env.addSource( new MySensorSource());

我们希望可以随机生成传感器数据，MySensorSource具体的代码实现如下：

package com.atguigu.apitest.source;
    import com.atguigu.apitest.beans.SensorReading;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.source.SourceFunction;
    
    import java.util.HashMap;
    import java.util.Random;
    
    public class SourceTest4_UDF {
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            // 从文件读取数据
            DataStream<SensorReading> dataStream = env.addSource( new MySensorSource() );
    
            // 打印输出
            dataStream.print();
    
            env.execute();
        }
    
        // 实现自定义的SourceFunction
        public static class MySensorSource implements SourceFunction<SensorReading>{
            // 定义一个标识位，用来控制数据的产生
            private boolean running = true;
    
            @Override
            public void run(SourceContext<SensorReading> ctx) throws Exception {
                // 定义一个随机数发生器
                Random random = new Random();
    
                // 设置10个传感器的初始温度
                HashMap<String, Double> sensorTempMap = new HashMap<>();
                for( int i = 0; i < 10; i++ ){
                    sensorTempMap.put("sensor_" + (i+1), 60 + random.nextGaussian() * 20);
                }
    
                while (running){
                    for( String sensorId: sensorTempMap.keySet() ){
                        // 在当前温度基础上随机波动
                        Double newtemp = sensorTempMap.get(sensorId) + random.nextGaussian();
                        sensorTempMap.put(sensorId, newtemp);
                        ctx.collect(new SensorReading(sensorId, System.currentTimeMillis(), newtemp));
                    }
                    // 控制输出频率
                    Thread.sleep(1000L);
                }
            }
    
            @Override
            public void cancel() {
                running = false;
            }
        }}

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70]: /images/20221024/ec040fbc9ca84dfdb91e929a4e55279a.png
[2021022023281646.png]: /images/20221024/b592598dd8e64cefb054f2b9881ae258.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1]: /images/20221024/4bec015fb68e4e408bb99d4c2ef9d63c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2]: /images/20221024/7a3ce3248de94d8e8bc15aafc7e2f6f3.png