Apache Flink -如何在启动时跳过除最新窗口之外的所有窗口

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 创建数据流
DataStream<Tuple2<String, Integer>> dataStream = env
  .socketTextStream("localhost", 9999)
  .flatMap(new Tokenizer());

// 定义窗口大小和滑动步长
Time windowSize = Time.seconds(10);
Time slideSize = Time.seconds(10);

// 使用滑动窗口，并跳过除最新窗口之外的所有窗口
dataStream
  .keyBy(0)
  .window(SlidingProcessingTimeWindows.of(windowSize, slideSize))
  .process(new WindowProcessFunction())
  .print();

// 执行任务
env.execute("Flink Window Skipping Example");

// 自定义Tokenizer函数用于将输入的字符串分割成单词
public class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
  @Override
  public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
    // 将字符串按空格分割成单词
    String[] words = value.toLowerCase().split("\\W+");

    // 输出每个单词的频率
    for (String word : words) {
      if (word.length() > 0) {
        out.collect(new Tuple2<>(word, 1));
      }
    }
  }
}

// 自定义WindowProcessFunction函数用于处理窗口数据
public class WindowProcessFunction extends ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String, TimeWindow> {
  @Override
  public void process(String key, Context context, Iterable<Tuple2<String, Integer>> input, Collector<Tuple2<String, Integer>> out) {
    // 处理最新窗口的数据
    for (Tuple2<String, Integer> item : input) {
      out.collect(item);
    }
  }
}