Flink1.12版本后建议使用assignTimestampsAndWatermarks(WatermarkStrategy)的方式生成watermark，之前使用的assignTimestampsAndWatermarks(AssignerWithPunctuatedWatermarks)以及assignTimestampsAndWatermarks(timestampAndWatermarkAssigner)均不再被推荐使用，这里是根据官网给出的小例子，仅供参考：
package demo;import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import java.time.Duration;public class WatermarkWCDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> inputStream = env.socketTextStream("localhost", 7777);SerializableTimestampAssigner<String> timestampAssigner = new SerializableTimestampAssigner<String>() {@Overridepublic long extractTimestamp(String element, long recordTimestamp) {String[] fields = element.split(",");Long aLong = new Long(fields[0]);return aLong * 1000L;}};SingleOutputStreamOperator<String> watermarkStream = inputStream.assignTimestampsAndWatermarks(WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(10)).withTimestampAssigner(timestampAssigner));watermarkStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {@Overridepublic void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {String[] fields = value.split(",");out.collect(new Tuple2<>(fields[1], 1));}}).keyBy(data -> data.f0).window(TumblingEventTimeWindows.of(Time.seconds(10))).sum(1).print();env.execute("run watermark wc");}
}
程序中，输入的nc数据形如：
1614909710,java
1614909701,python
1614909700,python
1614909702,python