在Flink SQL窗口中使用事件时间和延迟

DataStream<ApplicationMetric> metrics = env
  .addSource(new FlinkKinesisConsumer<>("applicationMetric", new SimpleStringSchema(), consumerConfig))
  .map(mapper)
  .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<>(Time.seconds(30)));  

打印为:

Table table = bsTableEnv.fromDataStream(dataset, "createdAt, name, duration, rowtime.rowtime");
Table t1 = bsTableEnv.sqlQuery("SELECT CAST((createdAt/1000) as TIMESTAMP) as ts, rowtime, name, duration " + table);
bsTableEnv.toAppendStream(t1, Row.class).print();