在Flink中，有三种迭代类型，一种对应DataStream，两种对应DataSet

DataStream Iteration

这里的iterationBody接收X的输入，并根据最后的closeWith完成一次迭代，另外的数据操作Z退出迭代。

由于DataStream有可能永远不停止，因此不能设置最大的迭代次数，必须显式的指定哪些流需要迭代，哪些流需要输出。

这时候，一般使用split transformation 或者 filter来实现。

下面定义了一个0-1000的字符流，每次迭代都减1，每个数直到小于等于1时输出（退出循环）：

DataStream<Long> someIntegers = env.generateSequence(0, 1000);

IterativeStream<Long> iteration = someIntegers.iterate();

DataStream<Long> minusOne = iteration.map(new MapFunction<Long, Long>() {
  @Override
  public Long map(Long value) throws Exception {
    return value - 1 ;
  }
});

DataStream<Long> stillGreaterThanZero = minusOne.filter(new FilterFunction<Long>() {
  @Override
  public boolean filter(Long value) throws Exception {
    return (value > 0);
  }
});

iteration.closeWith(stillGreaterThanZero);

DataStream<Long> lessThanZero = minusOne.filter(new FilterFunction<Long>() {
  @Override
  public boolean filter(Long value) throws Exception {
    return (value <= 0);
  }
});