- 不同类型的传感器数据频率不同，有的低频几分钟一个数，有的高频一秒几十个数，几百个数。传统的低频数据可以使用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();
}
}