Как создать экземпляр MapStateDescriptor в Fink для вычисления нескольких средних потоковых запросов?

Question

Я пытаюсь вычислить среднее значение температуры для 3 разных комнат, где в каждой комнате есть 3 датчика температуры.Я использую Flink (на Java).Сначала я делю датчики ключом, который является комнатой (A, B или C), и создаю им RichFlatMapFunction, который содержит MapState для сохранения температуры, пока у меня нет до 3 измерений.После трех измерений я вычисляю среднее значение.Чтобы использовать MapState, мне нужен MapStateDescriptor, который я не знаю, как правильно создать экземпляр.Может кто-то помочь мне с этим?Спасибо.

public class SensorsMultipleReadingMqttEdgentQEP2 {

    private boolean checkpointEnable = false;
    private long checkpointInterval = 10000;
    private CheckpointingMode checkpointMode = CheckpointingMode.EXACTLY_ONCE;

    public SensorsMultipleReadingMqttEdgentQEP2() throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);

        if (checkpointEnable) {
            env.enableCheckpointing(checkpointInterval, checkpointMode);
        }

        DataStream<MqttTemperature> temperatureStream01 = env.addSource(new TemperatureMqttConsumer("topic-edgent-01"));
        DataStream<MqttTemperature> temperatureStream02 = env.addSource(new TemperatureMqttConsumer("topic-edgent-02"));
        DataStream<MqttTemperature> temperatureStream03 = env.addSource(new TemperatureMqttConsumer("topic-edgent-03"));

        DataStream<Tuple2<String, Double>> averageStream01 = temperatureStream01.map(new SensorMatcher()).keyBy(0)
                .flatMap(new AverageTempMapper());
        DataStream<Tuple2<String, Double>> averageStream02 = temperatureStream02.map(new SensorMatcher()).keyBy(0)
                .flatMap(new AverageTempMapper());
        DataStream<Tuple2<String, Double>> averageStream03 = temperatureStream03.map(new SensorMatcher()).keyBy(0)
                .flatMap(new AverageTempMapper());

        DataStream<Tuple2<String, Double>> averageStreams = averageStream01.union(averageStream02)
                .union(averageStream03);

        averageStreams.print();
        env.execute("SensorsMultipleReadingMqttEdgentQEP");
    }

    public static class SensorMatcher implements MapFunction<MqttTemperature, Tuple2<String, MqttTemperature>> {

        private static final long serialVersionUID = 7035756567190539683L;

        @Override
        public Tuple2<String, MqttTemperature> map(MqttTemperature value) throws Exception {
            String key = "no-room";
            if (value.getId().equals(1) || value.getId().equals(2) || value.getId().equals(3)) {
                key = "room-A";
            } else if (value.getId().equals(4) || value.getId().equals(5) || value.getId().equals(6)) {
                key = "room-B";
            } else if (value.getId().equals(7) || value.getId().equals(8) || value.getId().equals(9)) {
                key = "room-C";
            } else {
                System.err.println("Sensor not defined in any room.");
            }
            return new Tuple2<>(key, value);
        }
    }

    public static class AverageTempMapper
            extends RichFlatMapFunction<Tuple2<String, MqttTemperature>, Tuple2<String, Double>> {

        private static final long serialVersionUID = -4780146677198295204L;
        private MapState<String, Tuple2<Integer, Double>> modelState;

        @Override
        public void open(Configuration parameters) throws Exception {
            TypeInformation<Tuple2<String, Tuple2<Integer, Double>>> typeInformation = TypeInformation
                    .of(new TypeHint<Tuple2<String, Tuple2<Integer, Double>>>() {
                    });

            // HOW TO INSTANTIATE THIS descriptor?
            MapStateDescriptor<String, Tuple2<Integer, Double>> descriptor = new MapStateDescriptor<>("modelState",
                    String.class, Tuple2.class);
            modelState = getRuntimeContext().getMapState(descriptor);
        }

        @Override
        public void flatMap(Tuple2<String, MqttTemperature> value, Collector<Tuple2<String, Double>> out)
                throws Exception {
            Double temp = null;
            Integer count = 0;
            if (modelState.contains(value.f0)) {
                count = modelState.get(value.f0).f0 + 1;
                temp = (modelState.get(value.f0).f1 + value.f1.getTemp());
            } else {
                count = 1;
                temp = value.f1.getTemp();
            }
            modelState.put(value.f0, Tuple2.of(count, temp));

            if (count >= 3) {
                out.collect(Tuple2.of("room", null));
            }
        }
    }
}

Answer 1

Чтобы определить MapStateDescriptor, вы можете сделать следующее:

MapStateDescriptor<String, Tuple2<Integer, Double>> modelState = new MapStateDescriptor<>(
    "modelState", 
    BasicTypeInfo.STRING_TYPE_INFO, 
    TupleTypeInfo.getBasicTupleTypeInfo(Integer.class, Double.class));
this.modelState = getRuntimeContext().getMapState(modelState);

Однако на самом деле нет необходимости использовать MapState в вашем случае.Поскольку поток уже имеет ключ, достаточно использовать ValueState.Код будет выглядеть следующим образом:

public static class AverageTempMapper extends RichFlatMapFunction<Tuple2<String, MqttTemperature>, Tuple2<String, Double>> {

    private static final long serialVersionUID = -4780146677198295204L;
    private ValueState<Tuple2<Integer, Double>> modelState;

    @Override
    public void open(Configuration parameters) {
        this.modelState = getRuntimeContext().getState(new ValueStateDescriptor<>("modelState", TupleTypeInfo.getBasicTupleTypeInfo(Integer.class, Double.class)));
    }

    @Override
    public void flatMap(Tuple2<String, MqttTemperature> value, Collector<Tuple2<String, Double>> out) throws Exception {
        Double temp;
        Integer count;
        if (modelState.value() != null) {
            Tuple2<Integer, Double> state = modelState.value();
            count = state.f0 + 1;
            temp = state.f1 + value.f1.getTemp();
        } else {
            count = 1;
            temp = value.f1.getTemp();
        }
        modelState.update(Tuple2.of(count, temp));

        if (count >= 3) {
            out.collect(Tuple2.of(value.f0, temp/count));
        }
    }
}

Answer 2

Я так и сделал, что считаю разумным.По крайней мере, это работает.Если у кого-то есть лучший подход для вычисления нескольких средних, не могли бы вы прокомментировать, пожалуйста?

public class SensorsMultipleReadingMqttEdgentQEP2 {

    private boolean checkpointEnable = false;
    private long checkpointInterval = 10000;
    private CheckpointingMode checkpointMode = CheckpointingMode.EXACTLY_ONCE;

    public SensorsMultipleReadingMqttEdgentQEP2() throws Exception {

        // Start streaming from fake data source sensors
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // obtain execution environment, run this example in "ingestion time"
        env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);

        if (checkpointEnable) {
            env.enableCheckpointing(checkpointInterval, checkpointMode);
        }

        DataStream<MqttTemperature> temperatureStream01 = env.addSource(new TemperatureMqttConsumer("topic-edgent-01"));
        DataStream<MqttTemperature> temperatureStream02 = env.addSource(new TemperatureMqttConsumer("topic-edgent-02"));
        DataStream<MqttTemperature> temperatureStream03 = env.addSource(new TemperatureMqttConsumer("topic-edgent-03"));

        DataStream<Tuple2<String, Double>> averageStream01 = temperatureStream01.map(new SensorMatcher()).keyBy(0)
                .flatMap(new AverageTempMapper());
        DataStream<Tuple2<String, Double>> averageStream02 = temperatureStream02.map(new SensorMatcher()).keyBy(0)
                .flatMap(new AverageTempMapper());
        DataStream<Tuple2<String, Double>> averageStream03 = temperatureStream03.map(new SensorMatcher()).keyBy(0)
                .flatMap(new AverageTempMapper());

        DataStream<Tuple2<String, Double>> averageStreams = averageStream01.union(averageStream02)
                .union(averageStream03);

        averageStreams.print();

        env.execute("SensorsMultipleReadingMqttEdgentQEP");
    }

    public static class SensorMatcher implements MapFunction<MqttTemperature, Tuple2<String, MqttTemperature>> {

        private static final long serialVersionUID = 7035756567190539683L;

        @Override
        public Tuple2<String, MqttTemperature> map(MqttTemperature value) throws Exception {
            String key = "no-room";
            if (value.getId().equals(1) || value.getId().equals(2) || value.getId().equals(3)) {
                key = "room-A";
            } else if (value.getId().equals(4) || value.getId().equals(5) || value.getId().equals(6)) {
                key = "room-B";
            } else if (value.getId().equals(7) || value.getId().equals(8) || value.getId().equals(9)) {
                key = "room-C";
            } else {
                System.err.println("Sensor not defined in any room.");
            }
            return new Tuple2<>(key, value);
        }
    }

    public static class AverageTempMapper
            extends RichFlatMapFunction<Tuple2<String, MqttTemperature>, Tuple2<String, Double>> {

        private static final long serialVersionUID = -4780146677198295204L;
        private MapState<String, Tuple2<Integer, Double>> modelState;
        private Integer threshold = 3;

        @Override
        public void open(Configuration parameters) throws Exception {
            TypeInformation<Tuple2<Integer, Double>> typeInformation = TypeInformation
                    .of(new TypeHint<Tuple2<Integer, Double>>() {
                    });

            MapStateDescriptor<String, Tuple2<Integer, Double>> descriptor = new MapStateDescriptor<String, Tuple2<Integer, Double>>(
                    "modelState", TypeInformation.of(String.class), typeInformation);
            modelState = getRuntimeContext().getMapState(descriptor);
        }

        @Override
        public void flatMap(Tuple2<String, MqttTemperature> value, Collector<Tuple2<String, Double>> out)
                throws Exception {
            Integer count = 0;
            Double temp = 0.0;

            if (modelState.contains(value.f0)) {
                // there is already a value on the state
                count = modelState.get(value.f0).f0 + 1;
                temp = modelState.get(value.f0).f1 + value.f1.getTemp();
                modelState.put(value.f0, Tuple2.of(1, value.f1.getTemp()));
            } else {
                // there is no value on the state
                count = 1;
                temp = value.f1.getTemp();
            }
            modelState.put(value.f0, Tuple2.of(count, temp));

            if (count >= threshold) {
                // only compute the average after the threshold
                out.collect(Tuple2.of(value.f0, temp / count));
                // clear the modelState value in order to compute new values next time
                modelState.put(value.f0, Tuple2.of(0, 0.0));
            }
        }
    }
}