Flink-算子（1）——DataSet

Posted 2023-05-17

参考技术A

fromCollection：从本地集合读取数据

例：

readTextFile：从文件中读取：

readTextFile可以对一个文件目录内的所有文件，包括所有子目录中的所有文件的遍历访问方式：

对于以下压缩类型，不需要指定任何额外的inputformat方法，flink可以自动识别并且解压。但是，压缩文件可能不会并行读取，可能是顺序读取的，这样可能会影响作业的可伸缩性。

因为Transform算子基于Source算子操作，所以首先构建Flink执行环境及Source算子，后续Transform算子操作基于此：

将DataSet中的每一个元素转换为另外一个元素：

将DataSet中的每一个元素转换为0...n个元素：

将一个分区中的元素转换为另一个元素：

过滤出来一些符合条件的元素，返回 boolean值为true 的元素：

可以对一个dataset或者一个group来进行聚合计算，最终 聚合成一个元素：

将一个dataset或者一个group 聚合成一个或多个元素 。
reduceGroup是reduce的一种优化方案；
它会先分组reduce，然后在做整体的reduce；这样做的好处就是可以减少网络IO：

选择具有最小值或最大值的 元素：

在数据集上进行聚合求 最值 （最大值、最小值）：

Aggregate只能作用于元组上

去除重复的数据：

取前N个数：

将两个DataSet按照一定条件连接到一起，形成新的DataSet：

左外连接,左边的Dataset中的每一个元素，去连接右边的元素

此外还有：

rightOuterJoin：右外连接,左边的Dataset中的每一个元素，去连接左边的元素

fullOuterJoin：全外连接,左右两边的元素，全部连接

下面以 leftOuterJoin 进行示例：

交叉操作，通过形成这个数据集和其他数据集的笛卡尔积，创建一个新的数据集

和join类似，但是这种交叉操作会产生笛卡尔积，在 数据比较大的时候，是非常消耗内存的操作：

联合操作，创建包含来自该数据集和其他数据集的元素的新数据集, 不会去重：

Flink也有数据倾斜的时候，比如当前有数据量大概10亿条数据需要处理，在处理过程中可能会发生如图所示的状况：

这个时候本来总体数据量只需要10分钟解决的问题，出现了数据倾斜，机器1上的任务需要4个小时才能完成，那么其他3台机器执行完毕也要等待机器1执行完毕后才算整体将任务完成；所以在实际的工作中，出现这种情况比较好的解决方案就是接下来要介绍的— rebalance （内部使用round robin方法将数据均匀打散。这对于数据倾斜时是很好的选择。）

按照指定的key进行hash分区：

根据指定的key对数据集进行范围分区：

根据指定的字段值进行分区的排序：

将数据输出到本地集合：

将数据输出到文件

Flink支持多种存储设备上的文件，包括本地文件，hdfs文件等

Flink支持多种文件的存储格式，包括text文件，CSV文件等

flink groupby keyby区别

1.groupby与keyby区别

spark中我们经常使用groupby算子对数据进行聚合。flink中，不仅有groupby算法，还有keyby算子，那么这两者的区别在哪里？
直接说结论：
groupby是用在DataSet系列API中，Table/SQL等操作也是使用groupby。
keyby是用在DataStream系列API中。

2.groupby简单实例

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class GroupBy 
    public static void groupbycode() throws Exception 
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataSet<String> text = env.fromElements("java python java python python c");
        DataSet<Tuple2<String, Integer>> dataSet = text.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() 
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception 
                for(String word: value.split(" ")) 
                    out.collect(new Tuple2<>(word, 1));
                
            
        );
        dataSet = dataSet.groupBy(0)
                .sum(1);
        dataSet.print();
    

    public static void main(String[] args) throws Exception 
        groupbycode();
    

上面可以认为是batch版的wordcount操作，对于DataSet使用的就是groupBy操作。

3.keyby简单实例

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class StreamWordCount 

    public static final class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> 
        @Override
        public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception 
            for(String word: s.split(" ")) 
                collector.collect(new Tuple2<String, Integer>(word, 1));
            
        
    

    public static void main(String[] args) throws Exception 
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<Tuple2<String, Integer>> dataStream = env
                .socketTextStream("localhost", 9999)
                .flatMap(new Splitter())
                .keyBy(value -> value.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(1)))
                .sum(1);

        dataStream.print();
        env.execute("Window WordCount");
    

上面是stream版的wordcount操作，对于DataStream数据，使用的则是keyby算子。