ClassifyData.Java

class ClassifyData extends RichFlatMapFunction<TestData, LabeledTestData>{

        ValueState<Integer> threshold;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            threshold = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("thresholdState", Integer.class));
        }

        @Override
        public void flatMap(TestData testData, Collector<LabeledTestData> collector) throws Exception {
            LabeledTestData labeledTestData = new LabeledTestData();
            labeledTestData.setKey(testData.getKey());
            labeledTestData.setValue(testData.getValue());
            labeledTestData.setCreatedAt(testData.getCreatedAt());
            String label = "UNCLASSIFIED";
            if(threshold.value() != null){
                label = (testData.getValue() > threshold.value()) ? "L1" : "L2";
            }
            labeledTestData.setLabel(label);
            collector.collect(labeledTestData);
        }
}