传感器采集的数据怎么处理、怎么存储【flink】【influxdb】_工控传感器数据缓存

Background

• 不同类型的传感器数据频率不同，低频的有的几分钟一个数，高频的有的一秒几十个数、几百个数。低频数据可以使用传统的mysql进行数据的存储。但数据频率比较高时，对程序的计算能力和数据存储能力要求较高，还好有现成的轮子可以直接拿来使用。
• 本文介绍高频数据流的实时计算和存储，应用场景选用风电塔筒提升监测为例。
• 之前的博客中也有探索，这里算是总结下吧，写出来是希望和大家交流，哪里有问题，多多指点哈。
• 这里给出源码【170-tower-lift-processor】，但是你拿到肯定是起不来的，这里分享出来主要是看大体流程哈。

数据处理流程

private static void execStreamJob() {
        log.info("****************** 获取 Flink 执行环境");
        log.info("");
        // 获取 Flink 任务执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);


        log.info("****************** 配置 RabbitMQ 数据源并解析");
        log.info("");
        // 解析原始数据，把流元素封装成 List<MultiDataEntity>
        SingleOutputStreamOperator<List<MultiDataEntity>> rawStream = env
                .setParallelism(1)
                // 配置 RabbitMQ 数据源
                .addSource(DefaultConfig.getRMQSource())
                // 统计数据频率
                .process(new PrintSpeedFunc())
                // 过滤掉为 null 和 whitespace 的值
                .filter(new NullFilter())
                // 解析数据
                .process(new RMQPayloadParser());


        log.info("****************** 原始数据存储【influx】");
        log.info("");
        // 存储原始数据到 influxDB 中
        rawStream
                .addSink(new RawInfluxSink());


        log.info("****************** 把需要进行特殊计算的类型分出来成一道侧流【分流】");
        log.info("");
        // 其他
        OutputTag<List<MultiDataEntity>> tag1 = new OutputTag<List<MultiDataEntity>>("stream-other") {
        };
        // 应变
        OutputTag<List<MultiDataEntity>> tag2 = new OutputTag<List<MultiDataEntity>>("stream-yb") {
        };
        // 风环境
        OutputTag<List<MultiDataEntity>> tag3 = new OutputTag<List<MultiDataEntity>>("stream-we") {
        };
        // 位移
        OutputTag<List<MultiDataEntity>> tag4 = new OutputTag<List<MultiDataEntity>>("stream-wy") {
        };
        // 振动
        OutputTag<List<MultiDataEntity>> tag5 = new OutputTag<List<MultiDataEntity>>("stream-zd") {
        };
        // 倾角
        OutputTag<List<MultiDataEntity>> tag6 = new OutputTag<List<MultiDataEntity>>("stream-qj") {
        };
        // 拆分流
        SingleOutputStreamOperator<List<MultiDataEntity>> splitStream = rawStream
                .process(new TagStreamProcessFunc(tag1, tag2, tag3, tag4, tag5, tag6));
        // 其它类型流
        DataStream<List<MultiDataEntity>> otherStream = splitStream.getSideOutput(tag1);
        // 应变类型流
        DataStream<List<MultiDataEntity>> ybStream = splitStream.getSideOutput(tag2);
        // 风环境类型流
        DataStream<List<MultiDataEntity>> weStream = splitStream.getSideOutput(tag3);
        // 位移类型流
        DataStream<List<MultiDataEntity>> wyStream = splitStream.getSideOutput(tag4);
        // 振动类型流
        DataStream<List<MultiDataEntity>> zdStream = splitStream.getSideOutput(tag5);
        // 倾角类型流
        DataStream<List<MultiDataEntity>> qjStream = splitStream.getSideOutput(tag6);


        log.info("****************** 振动原始数据推送 mqtt【振动】");
        log.info("");
        // 把振动传感器原始监测数据推送到 mqtt
        zdStream
                .addSink(new RawMqttSink());

        // 5 秒推送一个最大值和最小值
        SingleOutputStreamOperator<List<MultiDataEntity>> processedZDStream = zdStream
                .assignTimestampsAndWatermarks(new OriginFrequencyWatermark())
                .keyBy(new ProjectIdAndSensorTypeSelector())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new MaxAndMinProcessor());

        // 推送大屏展示
        processedZDStream
                .addSink(new MaxAndMinMqttSink());


        log.info("****************** 应变参考点缓存 redis【应变】");
        log.info("");
        // 把应变参考点，通讯编号为31、32的传感器数据缓存到 redis 中，实时更新
        ybStream
                .addSink(new YBToRedisSink());

        // 顶固的减去顶固的参考点（31）；地锚的减去地锚的参考点（32）；计算所得的数值设置成应变类型的【第6个指标】
        SingleOutputStreamOperator<List<MultiDataEntity>> ybPFStream = ybStream
                .process(new YBProcessor());

        // 其他传感器数据【10秒去除一个最大值和一个最小值然后返回一个瞬时值】只推送 不存储
        SingleOutputStreamOperator<List<MultiDataEntity>> ybPSStream = ybPFStream
                .assignTimestampsAndWatermarks(new OriginFrequencyWatermark())
                .keyBy(new ProjectIdAndSensorTypeSelector())
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .process(new RemoveMaxAndMinProcessor());
        // 推送大屏
        ybPSStream
                .addSink(new YBMqttSink());

        // 对应变类型流进行前十秒均值计算，计算所得的数值设置成应变类型的【第5个指标】
        SingleOutputStreamOperator<List<MultiDataEntity>> processedYBStream = ybPFStream
                .assignTimestampsAndWatermarks(new OriginFrequencyWatermark())
                .keyBy(new ProjectIdAndSensorTypeSelector())
                .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.milliseconds(500)))
                .apply(new TenSecMeanApplyFunc());


        log.info("****************** 风/叶夹角实时计算【风环境】");
        log.info("");
        // 对风环境类型流进行【风/叶夹角】实时计算
        SingleOutputStreamOperator<List<MultiDataEntity>> processedWEStream = weStream
                .process(new WEProcessor());


        log.info("****************** 位移类型数据处理【位移】");
        log.info("");
        // 计算当前轮毂实际高度
        SingleOutputStreamOperator<List<MultiDataEntity>> processedWYStream = wyStream
                .process(new WYProcessor());

        // 处理平整度数据（减去对应的复位值）
        wyStream
                .assignTimestampsAndWatermarks(new OriginFrequencyWatermark())
                .keyBy(new ProjectIdAndSensorTypeSelector())
                .window(TumblingEventTimeWindows.of(Time.seconds(2)))
                .process(new EvennessProcessor());


        log.info("****************** 倾角数据处理【倾角】");
        log.info("");
        // 1秒内去除一个最大值然后返回一个最大值
        SingleOutputStreamOperator<List<MultiDataEntity>> processedQJStream = qjStream
                .assignTimestampsAndWatermarks(new OriginFrequencyWatermark())
                .keyBy(new ProjectIdAndSensorTypeSelector())
                .window(TumblingEventTimeWindows.of(Time.seconds(1)))
                .process(new RemoveMaxProcessor());

        // 把处理后的倾角数据缓存到 redis 进行法兰倾斜计算
        processedQJStream
                .addSink(new QJRedisSink());


        log.info("****************** 合并所有侧流，存储并进行阈值告警判断【合并流】");
        log.info("");
        // 合并所有侧流
        DataStream<List<MultiDataEntity>> unionStream = zdStream
                .union(processedYBStream)
                .union(processedWEStream)
                .union(processedWYStream)
                .union(processedQJStream);


        log.info("****************** 计算结果存储【influxDB】");
        log.info("");
        // 把均值计算结果写入 influxDB
        unionStream.addSink(new MeanInfluxSink());


        log.info("****************** 【阈值告警判断】");
        log.info("");
        // 先推送实时数据，再进行阈值判断，告警信息推送 mqtt；同时把告警信息存储 redis，定时任务存储 MySQL
        unionStream.addSink(new WarnDetector());


        log.info("****************** 任务配置完毕，流计算开始 . . . ");
        log.info("");
        try {
            // 因为Flink是懒加载的，所以必须调用execute方法，上面的代码才会执行
            env.execute(DefaultConfig.getJobName());
        } catch (Exception e) {
            log.error("流计算任务执行失败！");
            log.info("errMsg: {}", e.getMessage());
            e.printStackTrace();
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181