请问flinksql如何控制结果输出的频率呢？

我们也经常有固定窗口定期触发的需求，但是添加以下参数并没有得到预期的效果（10秒的窗口，期待每秒都输出结果），是我的使用方法不对还是其他问题呢？多谢各位，下面是伪代码：

public class EarlyEmitter { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1);

EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlink StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

tEnv.getConfig().getConfiguration().setBoolean("table.exec.emit.early-fire.enabled", true); tEnv.getConfig().getConfiguration().setString("table.exec.emit.early-fire.delay", "1000 ms");

Table table = tEnv.fromDataStream( env.addSource(new SourceData()), "generate_time, name, city, id, event_time.proctime"); tEnv.createTemporaryView("person", table);

String emit = "SELECT name, COUNT(DISTINCT id)" + "FROM person " + "GROUP BY TUMBLE(event_time, interval '10' second), name";

Table result = tEnv.sqlQuery(emit); tEnv.toRetractStream(result, Row.class).print();

env.execute("IncrementalGrouping"); }

private static final class SourceData implements SourceFunction<Tuple4<Long, String, String, Long>> { @Override public void run(SourceContext<Tuple4<Long, String, String, Long>> ctx) throws Exception { while (true) { long time = System.currentTimeMillis(); ctx.collect(Tuple4.of(time, "flink", "bj", 1L)); } }

@Override public void cancel() {

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