登录
原创

【征文活动】Flink教程(二)Transformations 常用算子(上)

专栏Flink
发布于 2020-11-21 阅读 90
  • 后端
  • Java
  • Flink
  • 聚合
原创

一、前言

算子会将一个或多个DataStream转换成一个新的DataStream。

在工作中使用最多的也就这些DataStream转换算子,学好这些算子是入门Flink的必要。

好在Flink的某些算子和Java8的lambda函数很像,这便于理解。下面我会先介绍Java的语法,再介绍Flink的语法,由浅入深。

下面可以从图中看到DataStream和不同Stream之间,经过不同算子可以相互转换。

二、Map

2.1 Java Lambda的Map

Map对于Stream中包含的元素使用给定的转换函数进行转换操作,新生成的Stream只包含转换生成的元素。

简单来说,Map是映射的意思,可将T类型输入,转成R类型的输出。

如下示例是获取WordCount对象集合中的word字段集合。List<WordCount>转换成List<String>代码如下:

public class MapJavaDemo {

    public static void main(String[] args) {

        List<WordCount> wordCountList = new ArrayList<>();
        wordCountList.add(new WordCount("Flink", 3));
        wordCountList.add(new WordCount("Elasticsearch", 1));

        //将每个字段从WordCount类型转成String类型,生成List<String>
        List<String> words = wordCountList.stream()
                          .map(WordCount::getWord)
                          .collect(Collectors.toList());

        System.out.println(words);
    }
}

运行结果如下:

[Flink, Elasticsearch]

上面.map(WordCount::getWord)也可以写成.map(a -> a.getWord()),WordCount::getWord是一种简写方法罢了,a -> a.getWord()表达的更加直白。

2.2 Flink的Map

转换类型:DataStream → DataStream

  • 说明:读取一个元素并生成一个新的元素,例如
  • 举例:
输入 map转换 输出
1,2,3 乘以2 2,4,6
a,b,b 添加一个元素1,组成Tuple2<String, Integer> (a,1),(b,1),(b,1)

下面代码举例

  • 第一步:获取WordCount对象中word集合
  • 第二步:将每个count * 2
public class MapFlinkDemo {

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

        //创建环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        List<WordCount> wordCountList = new ArrayList<>();
        wordCountList.add(new WordCount("Flink", 3));
        wordCountList.add(new WordCount("Elasticsearch", 1));

        //设置数据源
        DataSource<WordCount> dataSource = env.fromCollection(wordCountList);

        //转换1:获取WordCount对象中word集合
        MapOperator<WordCount, String> words = dataSource.map(a -> a.getWord());

        //输出结果1
        words.print();

        //转换2:将每个count * 2
        MapOperator<WordCount, Object> doubleCount = dataSource.map(a -> {
            a.setCount(a.getCount() * 2);
            return a;
        });

        //输出结果2
        doubleCount.print();
    }
}

运行结果如下:

Flink
Elasticsearch
WordCount{word='Flink', count=6}
WordCount{word='Elasticsearch', count=2}

三、FlatMap

FlatMap也称扁平化map,它可把多个集合放到一个集合里。
举例说明:

  • WordCount程序中是把输入值,按照空格(" ")切分。
  • “i love you"和"you love me” -> [“i”,“love”,“you”]和[“you”,“love”,“me”]
  • Arrays.stream()将上面2个数组生成的2个Stream<String>流,
  • flatMap()把2个Stream<String>流合成1个Stream<String>

3.1 Java Lambda的FlatMap

public class FlatMapJavaDemo {

    public static void main(String[] args) {

        List<String> wordCountList = new ArrayList<>();
        wordCountList.add("i love you");
        wordCountList.add("you love me");

        Map<String, Long> collect = wordCountList.stream()

                //这一行会生成2个String[]:["i","love","you"]和["you","love","me"]
                .map(a -> a.split(" "))

                //Arrays.stream()将上面2个数组生成的2个Stream<String>流,
                //flatMap()把2个Stream<String>流合成1个Stream<String>
                .flatMap(Arrays::stream)
                
                //在1个Stream<String>上面根据单词聚合统计
                .collect(Collectors.groupingBy(w -> w, Collectors.counting()));
        
        System.out.println(collect);
    }
}

运行结果如下:

{love=2, me=1, i=1, you=2}

3.2 Flink的FlatMap

  • 转换类型:DataStream → DataStream
  • 说明:多组数据->生成多个流->合并成一个流
  • 举例:
输入 flatMap转换 输出
“I love coding”, “I love flink” 切分后,组成Tuple2<String, Integer> (flink,1)2个(love,1)2个(I,1)(coding,1)
public class FlatMapFlinkDemo {

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

        //创建环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        List<String> wordCountList = new ArrayList<>();
        wordCountList.add("i love you");
        wordCountList.add("you love me");

        //设置数据源
        DataSource<String> dataSource = env.fromCollection(wordCountList);

        //数据转换
        FlatMapOperator<String, String> wordStream = dataSource.flatMap(
                (FlatMapFunction<String, String>) (in, out) -> {
                    Arrays.stream(in.split(" ")).forEach(out::collect);
                }
        ).returns(Types.STRING);

        wordStream.print();
    }
}

四、Filter

4.1 Java Lambda的Filter

这里的filter方法接收的是Predicate<? super T> predicate,这个返回boolean类型。

public class FilterJavaDemo {

    public static void main(String[] args) {

        List<WordCount> wordCountList = new ArrayList<>();
        wordCountList.add(new WordCount("Flink", 3));
        wordCountList.add(new WordCount("Elasticsearch", 1));

        List<WordCount> collect = wordCountList.stream().filter(a -> a.getCount() > 1).collect(Collectors.toList());

        System.out.println(collect);
    }
}
[WordCount{word='Flink', count=3}]

4.2 Flink的Filter

  • 转换类型:DataStream → DataStream
  • 说明:该算子将按照条件对输入数据集进行筛选操作,将符合条件的数据集输出
  • 举例:
输入 flatMap转换 输出
1, 2, 3, 4, 5, 6 找到奇数 1,3,5
public class FilterFlinkDemo {

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

        //创建执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        List<WordCount> wordCountList = new ArrayList<>();
        wordCountList.add(new WordCount("Flink", 3));
        wordCountList.add(new WordCount("Elasticsearch", 1));

        DataSource<WordCount> source = env.fromCollection(wordCountList);

        FilterOperator<WordCount> filter = source.filter(a -> a.getCount() > 1);

        filter.print();
    }
}

运行结果如下:

WordCount{word='Flink', count=3}

五、KeyBy

在Flink中KeyBy就是分组,如果是DataSet用groupBy,是DataStream用keyBy。

5.1 Java lambda的groupingBy

先按第一个字段分组,再按第二个字段求和

public class GroupByJavaDemo {

    public static void main(String[] args) {

        List<WordCount> wordCountList = new ArrayList<>();
        wordCountList.add(new WordCount("Flink", 3));
        wordCountList.add(new WordCount("Flink", 2));
        wordCountList.add(new WordCount("Elasticsearch", 1));
        wordCountList.add(new WordCount("Elasticsearch", 5));

        //先按第一个字段分组,再按第二个字段求和
        Map<String, Integer> groupByThenSum = wordCountList.stream()
                .collect(
                        Collectors.groupingBy(
                                WordCount::getWord,
                                Collectors.summingInt(WordCount::getCount)
                        )
                );

        System.out.println(groupByThenSum);
    }
}
{Elasticsearch=6, Flink=5}

5.2 Flink的KeyBy

  • 转换类型:DataStream → KeyedStream

  • 说明:具有相同key的所有记录会分配给到同一分区,类似SQL的group by,在内部,keyBy()是使用hash分区实现

  • 如果是元祖,keyBy()里是下标,如keyBy(0), keyBy(0,1)

  • 如果是POJO,keyBy()里是字段名,如keyBy(“word”),如果根据多个字段分组,keyBy(“field1”, “field2”)

  • 举例:WordCount程序

public class StreamWordCountLambdaBetter {

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

        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //创建数据源
        DataStreamSource<String> lines = env.socketTextStream(BaseConstant.URL, BaseConstant.PORT);

        //将一行数据,按照空格分隔,将每个字符转成Tuple2.of(s, 1)
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = lines.flatMap(
                (String line, Collector<Tuple2<String, Integer>> out) -> {
                    Arrays.stream(line.split(" ")).forEach(thisWord -> {
                        out.collect(Tuple2.of(thisWord, 1));
                    });
                }
        ).returns(Types.TUPLE(Types.STRING, Types.INT));

        //统计操作
        SingleOutputStreamOperator<Tuple2<String, Integer>> sumResult = wordAndOne
                .keyBy(value -> value.f0)
                .sum(1);

        //sink
        sumResult.print();

        //执行
        env.execute("StreamWordCountLambdaBetter");
    }
}

六、Aggregations

6.1 Java Lambda的聚合函数

Java8中聚合函数非常简单,很容易理解和上手。

long count = pigs.stream().filter(a -> a.getAge() > 5).count();
System.out.println("age > 5的人数 = " + count);

//limit,取前几个
System.out.println("top2");
Stream<Pig> top2Pigs = pigs.stream().sorted(comparing(Pig::getAge)).limit(2);
top2Pigs.forEach(System.out::println);

int sumAge = pigs.stream().mapToInt(Pig::getAge).sum();
int maxAge = pigs.stream().mapToInt(Pig::getAge).max().getAsInt();
int minAge = pigs.stream().mapToInt(Pig::getAge).min().getAsInt();
double avgAge = pigs.stream().mapToInt(Pig::getAge).average().getAsDouble();

System.out.println("sumAge = " + sumAge);
System.out.println("maxAge = " + maxAge);
System.out.println("minAge = " + minAge);
System.out.println("avgAge = " + avgAge);

Optional<Pig> pigMaxAgeOptional = pigs.stream().collect(Collectors.maxBy(comparing(Pig::getAge)));
if (pigMaxAgeOptional.isPresent()){
    System.out.println("maxAge = " + pigMaxAgeOptional.get().getAge());
}

6.2 Flink中的Aggregations

Flink中聚合有如下
其中要注意:

  • min根据指定的字段取最小,它只返回最小的那个字段,而不是整个数据元素,对于其他的字段取了第一次取的值,不能保证每个字段的数值正确。
  • minBy根据指定字段取最小,返回的是整个元素。
keyedStream.sum(0);
keyedStream.sum(“key”);
keyedStream.min(0);
keyedStream.min(“key”);
keyedStream.max(0);
keyedStream.max(“key”);
keyedStream.minBy(0);
keyedStream.minBy(“key”);
keyedStream.maxBy(0);
keyedStream.maxBy(“key”);
public class StreamMinTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        List data = new ArrayList<Tuple3<String, String, Integer>>();
        data.add(new Tuple3<>("男", "老王", 80));
        data.add(new Tuple3<>("男", "小王", 25));
        data.add(new Tuple3<>("男", "老李", 85));
        data.add(new Tuple3<>("男", "小李", 20));

        DataStreamSource peoples = env.fromCollection(data);
		
		//求年龄最小的人
        SingleOutputStreamOperator minResult = peoples.keyBy(0).min(2);
        minResult.print();

        env.execute("StreamMinTest");
    }
}

返回结果如下:

1> (男,老王,80)
1> (男,老王,25)
1> (男,老王,25)#年龄算对了,但是别的字段还是第一个老王
1> (男,老王,20)#年龄最小的居然还是第一个老王?

七、总结

本篇博客只是总结了部分Java8和Flink的算子,他们在Java和Flink上有共性,所以拿来一起讲,
这个是网上别的Flink博客所没有的。
感觉学习还是温故而知新,不能停下学习的脚步,也得时不时的回顾以前的知识,才能有新的收获。
感觉凑够500字了,下一篇博客得讲讲Transformations 常用算子中的Reduce、Fold、Union、Connect、Split & Select。

评论区

菜鸟

0

0

0