在Apache Flink的表Api中从窗口聚合中获取部分结果

// 导入所需的类
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.types.Row;

// 创建流处理环境和表环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

// 定义自定义聚合函数
class SumAggFunction extends AggregateFunction<Long, SumAggFunction.SumAccumulator> {
    public static class SumAccumulator {
        public long sum = 0L;
    }

    @Override
    public SumAccumulator createAccumulator() {
        return new SumAccumulator();
    }

    @Override
    public Long getValue(SumAccumulator accumulator) {
        return accumulator.sum;
    }

    public void accumulate(SumAccumulator accumulator, Long value) {
        accumulator.sum += value;
    }
}

// 注册自定义聚合函数
tEnv.registerFunction("sumAgg", new SumAggFunction());

// 创建输入流表
tEnv.executeSql("CREATE TABLE input_table (name STRING, value BIGINT, event_time TIMESTAMP(3)) " +
        "WITH ('connector' = 'kafka', 'topic' = 'input_topic', 'properties.bootstrap.servers' = 'localhost:9092', " +
        "'format' = 'json', 'json.fail-on-missing-field' = 'false')");

// 执行窗口聚合操作
Table resultTable = tEnv.sqlQuery("SELECT name, TUMBLE_START(event_time, INTERVAL '1' HOUR) as window_start, " +
        "TUMBLE_END(event_time, INTERVAL '1' HOUR) as window_end, sumAgg(value) as sum_value " +
        "FROM input_table " +
        "GROUP BY name, TUMBLE(event_time, INTERVAL '1' HOUR)");

// 将结果表转换为流并打印结果
tEnv.toAppendStream(resultTable, Row.class).print();

// 提交作业并执行
env.execute();