Flink SQL是否支持Count Window函数？

据我所知目前暂时没有增加 count window 的打算，以后可能会在最新的 Window TVF 里添加 count window tvf。

不建议在 SQL 中自行实现 count window，因为 SQL 添加 window 较为复杂。但可以考虑先将 SQL 转为

datastream，用 datastream 的 count window 之后再将 datastream 转回 SQL。

EnvironmentSettings settings = EnvironmentSettings.newInstance().

inStreamingMode().build();

StreamTableEnvironment tEnv =

StreamTableEnvironment.create(

StreamExecutionEnvironment.getExecutionEnvironment(), settings);

tEnv.executeSql(

"CREATE TABLE T ( a INT, b INT, key AS abs(a) % 3, val AS abs(b) % 3 ) WITH

( 'connector' = 'datagen' )");

Table table = tEnv.sqlQuery("SELECT key, val FROM T");

DataStream dataStream = tEnv.toDataStream(table);

DataStream<Tuple2<Integer, Integer>> summedStream =

dataStream

.keyBy(row -> (int) row.getField(0))

.countWindow(100)

.apply(

(WindowFunction<

Row,

Tuple2<Integer, Integer>,

Integer,

GlobalWindow>)

(key, window, input, out) -> {

int sum = 0;

for (Row row : input) {

Integer field = (Integer) row.getField(1);

if (field != null) {

sum += field;

}

}

out.collect(Tuple2.of(key, sum));

})

.returns(

new TupleTypeInfo<>(

BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO));

Table summedTable = tEnv.fromDataStream(summedStream);

tEnv.registerTable("S", summedTable);

tEnv.executeSql("SELECT f0, f1 FROM S").print();

*来自志愿者整理的FLINK邮件归档