传感器采集的数据怎么处理、怎么存储【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         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();
        }
    }