如何在flink中实现KGroupTable用例

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.api.common.typeinfo.Types;

public class KGroupTableExample {
    public static void main(String[] args) throws Exception {
        // 设置 Flink 流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 定义数据源
        DataStream<Row> clickStream = env.fromElements(
            Row.of("user1", "2023-10-01 10:00:00"),
            Row.of("user2", "2023-10-01 10:05:00"),
            Row.of("user1", "2023-10-01 10:10:00"),
            Row.of("user3", "2023-10-01 10:15:00"),
            Row.of("user2", "2023-10-01 10:20:00")
        ).returns(Types.ROW(Types.STRING, Types.STRING));

        // 将 DataStream 转换为 Table
        Table clickTable = tableEnv.fromDataStream(clickStream, $("userId"), $("clickTime"));
        
        // 注册表
        tableEnv.createTemporaryView("ClickTable", clickTable);
    }
}

import static org.apache.flink.table.api.Expressions.$;

public class KGroupTableExample {
    public static void main(String[] args) throws Exception {
        // 设置 Flink 流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 定义数据源
        DataStream<Row> clickStream = env.fromElements(
            Row.of("user1", "2023-10-01 10:00:00"),
            Row.of("user2", "2023-10-01 10:05:00"),
            Row.of("user1", "2023-10-01 10:10:00"),
            Row.of("user3", "2023-10-01 10:15:00"),
            Row.of("user2", "2023-10-01 10:20:00")
        ).returns(Types.ROW(Types.STRING, Types.STRING));

        // 将 DataStream 转换为 Table
        Table clickTable = tableEnv.fromDataStream(clickStream, $("userId"), $("clickTime"));
        
        // 注册表
        tableEnv.createTemporaryView("ClickTable", clickTable);

        // 使用 KGroupTable 进行分组和聚合
        Table resultTable = tableEnv.from("ClickTable")
            .groupBy($("userId"))
            .select($("userId"), $("userId").count().as("clickCount"));

        // 将结果表转换为 DataStream 并打印
        DataStream<Row> resultStream = tableEnv.toChangelogStream(resultTable);
        resultStream.print();

        // 执行 Flink 作业
        env.execute("KGroupTable Example");
    }
}