private static DataStream<Tuple2<Integer, Integer>> getFirstTupleOutput(DataStream<MyInputPOJO> inputPOJOs, StreamTableEnvironment tableEnv) {

        DataStream<Integer> projectedTuples =
                inputPOJOs
                        .map(x -> x.myIntField)
                        .returns(Types.INT);

        Table projectedTupleTable = tableEnv.fromDataStream(projectedTuples,
                Schema.newBuilder()
                        .column("f0", "INT")
                        .columnByMetadata("rowtime", "TIMESTAMP_LTZ(3)")
                        .watermark("rowtime", "SOURCE_WATERMARK()")
                        .build());

        Table projectedCountsTable = tableEnv.sqlQuery(
                "SELECT f0, COUNT(*) AS f3, window_time AS rowtime " +
                        "FROM TABLE(TUMBLE(TABLE " + projectedTupleTable + ",  DESCRIPTOR(rowtime), INTERVAL '60' MINUTE)) " +
                        "GROUP BY f0, window_start, window_end, window_time");


        return tableEnv.toDataStream(projectedCountsTable)
                .map((MapFunction<Row, Tuple2<Integer, Integer>>) row -> new Tuple2<>((Integer) row.getField("f0"), ((Long) row.getField("f3")).intValue()))
                .returns(Types.TUPLE(Types.INT, Types.INT))
                .name("firstTupleOutput");

    }


private static DataStream<Tuple2<Integer, Integer>> getSecondTupleOutput(DataStream<Tuple2<Integer, Integer>> firstTupleOutput, StreamTableEnvironment tableEnv) {

        Table firstTupleOutputTable = tableEnv.fromDataStream(firstTupleOutput,
                Schema.newBuilder()
                        .column("f0", "INTEGER")
                        .column("f1", "INTEGER")
                        .columnByMetadata("rowtime", "TIMESTAMP_LTZ(3)")
                        .watermark("rowtime", "SOURCE_WATERMARK()")
                        .build());

        Table secondCountsTable = tableEnv.sqlQuery(
                "SELECT f0, SUM(f1) AS sum_f1, window_time AS rowtime " +
                        "FROM TABLE(HOP(TABLE " + firstTupleOutputTable + ",  DESCRIPTOR(rowtime), INTERVAL '1' HOUR, INTERVAL '1' DAY)) " +
                        "GROUP BY f0, window_start, window_end, window_time");


        return tableEnv.toDataStream(secondCountsTable)
                .map((MapFunction<Row, Tuple2<Integer, Integer>>) row -> new Tuple2<>((Integer) row.getField("f0"), (Integer) row.getField("sum_f1")))
                .returns(Types.TUPLE(Types.INT, Types.INT))
                .name("secondTupleOutput");
    }