1、开发算法阈值算法块

package com.censoft.flink.transform;

import com.censoft.flink.domain.AlgorithmPushDto;

import com.censoft.flink.domain.AlgorithmSceneBasePo;

import com.censoft.flink.domain.AlgorithmScenePiecePo;

import com.censoft.flink.domain.AlgorithmScenePieceVariablePo;

import com.censoft.flink.exception.ParameterTransformException;

import java.util.List;

import java.util.Optional;

import java.util.concurrent.ConcurrentHashMap;

import java.util.stream.Collectors;

/**

 * @author 周昊

 * @desc 报警阈值筛选

 */

public class AlarmThresholdFunction implements AlgorithmBaseFilterFunction {

    //threshold 阈值百分比（整数）

    private final Integer threshold;

    //对应报警间隔 参数KEY

    private static String thresholdKey = "threshold";

    public AlarmThresholdFunction(Integer threshold) {

        this.threshold = threshold;

    }

    public static AlarmThresholdFunction getFilterFunction(AlgorithmSceneBasePo algorithmSceneBasePo, AlgorithmScenePiecePo algorithmScenePiecePo) {

        //参数准备

        Optional<AlgorithmScenePieceVariablePo> variablePo = algorithmScenePiecePo.getVariablePos()

                .stream()

                .filter(po -> thresholdKey.equals(po.getVariableKey()))

                .findFirst();

        //判断参数是否存在，如果不存在抛出异常

        if (!variablePo.isPresent()) {

            throw new ParameterTransformException();

        }

        Integer threshold = Integer.valueOf(variablePo.get().getVariableValue());

        return new AlarmThresholdFunction(threshold);

    }

    @Override

    public boolean filter(AlgorithmPushDto algorithmPushDto) {

        return algorithmPushDto.getResult()

                .stream()

                .map(po -> {

                    String label = po.getLabel();

                    if (label.contains("_")) {

                        String thresholdString = label.substring(label.indexOf("_") + 1);

                        return Integer.valueOf(thresholdString);

                    } else {

                        return null;

                    }

                }).filter(str -> str != null)

                .filter(value -> value >= threshold)

                .findFirst().isPresent();

    }

}