Flink之DataSet迭代计算

短命女 2022-10-23 08:20 189阅读 0赞

### 目录 ###

*   *  （1）迭代计算分类与原理
     *  （2）全量迭代计算详解
     *   *  （2.1）案例分析
         *  （2.2）案例实战
     *  （3）增量迭代计算详解

## （1）迭代计算分类与原理 ##

迭代计算在批量数据处理过程中的应用非常广泛，如常用的机器学习算法Kmeans、逻辑回归，以及图形计算等，都会用到迭代计算。DataSet API对迭代计算功能的支持相对比较完善，在性能上比较其他分布式计算框架也具有非常高的优势。目前Flink中的迭代计算种类有两种模式，分别是**Bulk Iteration （全量迭代计算）和 Delt Iteration（增量迭代计算）**

什么是迭代运算？  
所谓迭代运算，就是给定一个初值，用所给的算法公式计算初值得到一个中间结果，然后将中间结果作为输入参数进行反复计算，在满足一定条件的时候得到计算结果。

## （2）全量迭代计算详解 ##

Bulk Iteration  
这种迭代方式称为全量迭代，它会将整个数据输入，经过一定的迭代次数  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]  
**全量迭代计算，一共有几个步骤:**

1.  首先初始化数据，可以通过从DataSource算子中获取，也可以从其他转化Operators中接入
2.  其次定义Step Function，并在每一步迭代过程使用Step Function，结合数据集以及上一次迭代计算的Solution数据集，进行本次迭代计算。
3.  每一次迭代过程中Step Function输出的结果，被称为Next Partital Solution数据集，该结果会作为下一次迭代计算的输入数据集。
4.  最后一次迭代计算的结果输出，可以通过DataSink输出，或者接入到下一个0perators中。

**迭代终止的条件有两种，分别为达到最大迭代次数或者符合自定义聚合器收敛条件:**

1.  最大迭代次数:指定迭代的最大次数，当计算次数超过该设定值是，终止迭代
2.  自定义收敛条件:用户自定义的聚合器和收敛条件，例如终止条件设定为当Sum统计结果小于零则终止，否则继续迭代。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 1]

### （2.1）案例分析 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 2]

### （2.2）案例实战 ###

全量迭代计算通过使用DataSet的iterate()方法调用:

package com.aikfk.flink.dataset.iteration;
    
    import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.api.java.DataSet;
    import org.apache.flink.api.java.ExecutionEnvironment;
    import org.apache.flink.api.java.operators.IterativeDataSet;
    
    /** * @author ：caizhengjie * @description：TODO * @date ：2021/3/9 5:59 下午 */
    public class PiIterator { 
        public static void main(String[] args) throws Exception { 
            // 准备环境
            ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    
            // 创建初始IterativeDataSet
            IterativeDataSet<Integer> initial = env.fromElements(0).iterate(10000);
    
            DataSet<Integer> iteration = initial.map(new MapFunction<Integer, Integer>() { 
                @Override
                public Integer map(Integer integer) throws Exception { 
                    double x = Math.random();
                    double y = Math.random();
                    return integer + ((x * x + y * y <= 1 ) ? 1 : 0);
                }
            });
    
            // 计算出点的距离小于一的个数
            DataSet<Integer> count = initial.closeWith(iteration);
    
            // 求出PI
            DataSet<Double> result = count.map(new MapFunction<Integer, Double>() { 
                @Override
                public Double map(Integer count) throws Exception { 
                    return count / (double) 10000 * 4;
                }
            });
    
            result.print();
    
            /** * 3.146 */
        }
    }

## （3）增量迭代计算详解 ##

增量迭代是通过部分计算取代全量计算，在计算过程中会将数据集分为热点数据和非热点数据集，每次迭代计算会针对热点数据展开，这种模式适合用于数据量比较大的计算场景，不需要对全部的数据集进行计算，所以在性能和速度上都会有很大的提升。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 3]

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221023/9899fe157c05443da9f0b51a3b5a5282.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221023/dca8b24cac33487fa86018828f0d619d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221023/5322c8e637c142188292d1360b3e2949.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221023/e65b74023a1a4762a655f1ec30faee8b.png