ClassifierJob.java

public class TestCheckpointJob {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

        Properties kafkaConsumerProperties = new Properties();
        kafkaConsumerProperties.setProperty("bootstrap.servers", "localhost:9092");
        kafkaConsumerProperties.setProperty("group.id", "test_group_id");

        ObjectMapper objectMapper = new ObjectMapper();
        FlinkKafkaConsumer010<String> kafkaConsumer010 = new FlinkKafkaConsumer010<>("test_topic", new SimpleStringSchema(), kafkaConsumerProperties);
        DataStream<String> kafkaSource =  env.addSource(kafkaConsumer010).name("kafka_source").uid("kafka_source");
        DataStream<TestData> aggregatedStream = kafkaSource
                .map(row -> objectMapper.readValue(row, TestData.class))
                .keyBy(TestData::getKey)
                .timeWindow(Time.hours(1))
                .reduce((rowA, rowB) -> {
                   TestData result = new TestData();
                    result.setKey(rowA.getKey());
                    result.setValue(rowA.getValue() + rowB.getValue());
                    result.setCreatedAt(System.currentTimeMillis());
                    return result;
                }).name("aggregate_stream").uid("aggregate_stream");

        DataStream<LabeledTestData> labeledTestDataDataStream =  aggregatedStream.keyBy(TestData::getKey).flatMap(new ClassifyData()).name("classify_data").uid("classify_data");
        labeledTestDataDataStream.map(row -> objectMapper.writeValueAsString(row)).print();
        env.execute();
    }
}