java Flink（七）从集合读取数据

矫情吗；* 2022-11-12 14:55 42阅读 0赞

今天继续学习Flink流处理的集合处理

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70][]

上边是我们创建的传感器对象，下边是Flink流处理程序

SensorReading:

package 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 +
                    '}';
        }
    }

SourceTest1\_Collection

package source;
    
    import beans.SensorReading;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    import java.util.Arrays;
    import java.util.List;
    
    public class SourceTest1_Collection {
        public static void main(String[] args) throws Exception {
            //创建执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //设置并行调度
            //env.setParallelism(1);
            //从集合中读取数据
            DataStream<SensorReading> dataStream = env.fromCollection(Arrays.asList(new SensorReading("sensor_1", 1537718199L, 35.8),
                    new SensorReading("sensor_6", 1547718201L, 15.4),
                    new SensorReading("sensor_7", 1547718202L, 6.7),
                    new SensorReading("sensor_10", 1547718205L, 38.1)));
    
            DataStream<Integer> integerDataStream = env.fromElements(1, 2, 4, 67, 189);
            //打印输出
            dataStream.print("data");
            integerDataStream.print("int");
            //执行 Flink的jobName
            env.execute("SensorReading");
        }
    }

查看执行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 1][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70]: /images/20221022/d85f7b0de76a468aaa43cdc137d798ed.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20210324150849485.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzcxNTY3,size_16,color_FFFFFF,t_70