flink Iterate迭代基本概念

基本概念：在流中創建“反饋（feedback）”循環，通過將一個算子的輸出重定向到某個先前的算子。這對於定義不斷更新模型的算法特別有用。

迭代的數據流向：DataStream → IterativeStream → DataStream

以下代碼以流開始並連續應用迭代體。大於0的元素將被發送回反饋（feedback）通道，繼續迭代，其余元素將向下游轉發，離開迭代。

IterativeStream<Long> iteration = initialStream.iterate();
DataStream<Long> iterationBody = iteration.map (/*do something*/);
DataStream<Long> feedback = iterationBody.filter(new FilterFunction<Long>(){
    @Override
    public boolean filter(Long value) throws Exception {
        return value > 0;
    }
});
//這里設置feedback這個數據流是被反饋的通道，只要是value>0的數據都會被重新迭代計算。
iteration.closeWith(feedback);
DataStream<Long> output = iterationBody.filter(new FilterFunction<Long>(){
    @Override
    public boolean filter(Long value) throws Exception {
        return value <= 0;
    }
});