DataStream

Posted 2022-09-12 JiaXingNashishua

目录

 5.3.2 聚合算子（Aggregation）

5.3.3 用户自定义函数（UDF）

---

3. 扁平映射（flatMap）

flatMap 操作又称为扁平映射，主要是将数据流中的整体（一般是集合类型）拆分成一个 一个的个体使用。消费一个元素，可以产生 0 到多个元素。flatMap 可以认为是“扁平化”（flatten） 和“映射”（map）两步操作的结合，也就是先按照某种规则对数据进行打散拆分，再对拆分 后的元素做转换处理。我们此前 WordCount 程序的第一步分词操作，就用到了flatMap。

 

同 map 一样，flatMap 也可以使用 Lambda 表达式或者 FlatMapFunction 接口实现类的方式 来进行传参，返回值类型取决于所传参数的具体逻辑，可以与原数据流相同，也可以不同。

flatMap 操作会应用在每一个输入事件上面， FlatMapFunction 接口中定义了 flatMap 方法， 用户可以重写这个方法，在这个方法中对输入数据进行处理，并决定是返回 0 个、1 个或多个 结果数据。因此 flatMap 并没有直接定义返回值类型，而是通过一个“收集器”（Collector）来 指定输出。希望输出结果时，只要调用收集器的.collect()方法就可以了；这个方法可以多次调 用，也可以不调用。

所以 flatMap 方法也可以实现 map 方法和 filter 方法的功能，当返回结果 是 0 个的时候，就相当于对数据进行了过滤，当返回结果是 1 个的时候，相当于对数据进行了 简单的转换操作。

代码实现：

package com.atguigu.chapter05;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class TransFromFlatMapTest 
    //1.实现自定义的FlatMapFunction接口
    public static class MyFlatMap implements FlatMapFunction<Event,String> 
        @Override
        public void flatMap(Event event, Collector<String> collector) throws Exception 
            collector.collect(event.user);
            collector.collect(event.url);
            collector.collect(event.timestamp.toString());
        
    
    public static void main(String[] args) throws Exception 
        StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //从元素中读取数据
        DataStreamSource<Event> stream=env.fromElements(
                new Event("mary","./home",1000L),
                new Event("Bob","./cart",2000L),
                new Event("Alice","./prod?id=100",3000L)
        );
        /*

        SingleOutputStreamOperator<String> result = stream.flatMap(new MyFlatMap());

        result.print();

         */

        //2.直接传入Lambda表达式(这里注意，要显示返回数据类型，否则报错)
        stream.flatMap((Event event,Collector<String> out)-> 
            out.collect(event.user);
            out.collect(event.url);
            out.collect(event.timestamp.toString());
        ).returns(new TypeHint<String>() )
                .print();

        env.execute();
    

 5.3.2 聚合算子（Aggregation）

直观上看，基本转换算子确实是在“转换”——因为它们都是基于当前数据，去做了处理 和输出。而在实际应用中，我们往往需要对大量的数据进行统计或整合，从而提炼出更有用的 信息。比如之前 word count 程序中，要对每个词出现的频次进行叠加统计。这种操作，计算 的结果不仅依赖当前数据，还跟之前的数据有关，相当于要把所有数据聚在一起进行汇总合并 ——这就是所谓的“聚合”（Aggregation），也对应着 MapReduce 中的 reduce 操作。

1. 按键分区（keyBy）

对于 Flink 而言，DataStream 是没有直接进行聚合的 API 的。因为我们对海量数据做聚合 肯定要进行分区并行处理，这样才能提高效率。所以在 Flink 中，要做聚合，需要先进行分区； 这个操作就是通过 keyBy 来完成的。

keyBy 是聚合前必须要用到的一个算子。keyBy 通过指定键（key），可以将一条流从逻辑 上划分成不同的分区（partitions）。这里所说的分区，其实就是并行处理的子任务，也就对应 着任务槽（task slot）。

基于不同的 key，流中的数据将被分配到不同的分区中去，如图  所示；这样一来，所有具有相同的 key 的数据，都将被发往同一个分区，那么下一步算子操作就将会在同一个 slot中进行处理了。

 

 在内部，是通过计算 key 的哈希值（hash code），对分区数进行取模运算来实现的。所以 这里 key 如果是 POJO 的话，必须要重写 hashCode()方法。 keyBy()方法需要传入一个参数，这个参数指定了一个或一组 key。有很多不同的方法来指 定 key：比如对于 Tuple 数据类型，可以指定字段的位置或者多个位置的组合；对于 POJO 类 型，可以指定字段的名称（String）；另外，还可以传入 Lambda 表达式或者实现一个键选择器 （KeySelector），用于说明从数据中提取 key 的逻辑。

import org.apache.flink.api.java.functions.KeySelector; 
import org.apache.flink.streaming.api.datastream.DataStreamSource; 
import org.apache.flink.streaming.api.datastream.KeyedStream; 
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; 
 
public class TransKeyByTest  
 public static void main(String[] args) throws Exception  
 StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment(); 
 env.setParallelism(1); 
 
 DataStreamSource<Event> stream = env.fromElements( 
 new Event("Mary", "./home", 1000L), 
 new Event("Bob", "./cart", 2000L) 
 ); 
 
 // 使用 Lambda 表达式 
 KeyedStream<Event, String> keyedStream = stream.keyBy(e -> e.user); 
 
 // 使用匿名类实现 KeySelector 
 KeyedStream<Event, String> keyedStream1 = stream.keyBy(new 

KeySelector<Event, String>()  
 @Override 
 public String getKey(Event e) throws Exception  
 return e.user; 
  
 ); 
 
 env.execute(); 
  
 

需要注意的是，keyBy 得到的结果将不再是 DataStream，而是会将 DataStream 转换为KeyedStream。KeyedStream 可以认为是“分区流”或者“键控流”，它是对 DataStream 按照key 的一个逻辑分区，所以泛型有两个类型：除去当前流中的元素类型外，还需要指定 key 的 类型。 KeyedStream 也继承自 DataStream，所以基于它的操作也都归属于 DataStream API。但它 跟之前的转换操作得到的 SingleOutputStreamOperator 不同，只是一个流的分区操作，并不是 一个转换算子。KeyedStream 是一个非常重要的数据结构，只有基于它才可以做后续的聚合操作（比如 sum，reduce）；而且它可以将当前算子任务的状态（state）也按照 key 进行划分、限 定为仅对当前 key 有效。

2. 简单聚合

有了按键分区的数据流 KeyedStream，我们就可以基于它进行聚合操作了。Flink 为我们 内置实现了一些最基本、最简单的聚合 API，主要有以下几种：

⚫ sum()：在输入流上，对指定的字段做叠加求和的操作。

⚫ min()：在输入流上，对指定的字段求最小值。

⚫ max()：在输入流上，对指定的字段求最大值。

⚫ minBy()：与 min()类似，在输入流上针对指定字段求最小值。不同的是，min()只计 算指定字段的最小值，其他字段会保留最初第一个数据的值；而 minBy()则会返回包 含字段最小值的整条数据。

⚫ maxBy()：与 max()类似，在输入流上针对指定字段求最大值。两者区别与min()/minBy()完全一致。

简单聚合算子使用非常方便，语义也非常明确。这些聚合方法调用时，也需要传入参数； 但并不像基本转换算子那样需要实现自定义函数，只要说明聚合指定的字段就可以了。指定字 段的方式有两种：指定位置，和指定名称。

对于元组类型的数据，同样也可以使用这两种方式来指定字段。需要注意的是，元组中字 段的名称，是以 f0、f1、f2、…来命名的。

package com.atguigu.chapter05;

import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

//单聚合测试
public class TransFromSimpleAggTest 
    public static void main(String[] args) throws Exception 
        StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> stream=env.fromElements(

                new Event("Bob","./cart",2000L),
                new Event("Alice","./prod?id=100",3000L),
                new Event("Bob","./prod?id=1",3300L),
                new Event("Bob","./home",3500L),
                new Event("Bob","./prod?id=3",4200L),
                new Event("Bob","./prod?id=2",3800L)
        );

        /*//按键分组之后进行聚合，提取当前用户最近一次访问数据
        stream.keyBy(new KeySelector<Event,String>() 
            @Override
            public String getKey(Event event) throws Exception 
                return event.user;
            
        ).max("timestamp").print("max: ");*/


        //Lambda表达式
        
        //maxBy：返回包含字段最大值的整条数据
        
        stream.keyBy(data -> data.user).maxBy("timestamp").print("maxBy: ");



        env.execute();
    

简单聚合算子返回的，同样是一个 SingleOutputStreamOperator，也就是从 KeyedStream 又 转换成了常规的 DataStream。所以可以这样理解：keyBy 和聚合是成对出现的，先分区、后聚合，得到的依然是一个 DataStream。而且经过简单聚合之后的数据流，元素的数据类型保持不变。

一个聚合算子，会为每一个key保存一个聚合的值，在Flink中我们把它叫作“状态”（state）。 所以每当有一个新的数据输入，算子就会更新保存的聚合结果，并发送一个带有更新后聚合值 的事件到下游算子。对于无界流来说，这些状态是永远不会被清除的，所以我们使用聚合算子， 应该只用在含有有限个 key 的数据流上。

3. 归约聚合（reduce）

educe 算子就是一个一般化的聚合 统计操作了。从大名鼎鼎的 MapReduce 开始，我们对 reduce 操作就不陌生：它可以对已有的 数据进行归约处理，把每一个新输入的数据和当前已经归约出来的值，再做一个聚合计算。

与简单聚合类似，reduce 操作也会将 KeyedStream 转换为 DataStream。它不会改变流的元 素数据类型，所以输出类型和输入类型是一样的。

package com.atguigu.chapter05;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class TransFromReduceTest 
    public static void main(String[] args) throws Exception
        StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> stream=env.fromElements(

                new Event("Bob","./cart",2000L),
                new Event("Alice","./prod?id=100",3000L),
                new Event("Bob","./prod?id=1",3300L),
                new Event("Bob","./home",3500L),
                new Event("Bob","./prod?id=3",4200L),
                new Event("Bob","./prod?id=2",3800L)
        );

        //1.统计每个用户的访问频次
        SingleOutputStreamOperator<Tuple2<String, Long>> clicksByUser = stream.map(new MapFunction<Event, Tuple2<String, Long>>() 
                    @Override
                    public Tuple2<String, Long> map(Event event) throws Exception 
                        return Tuple2.of(event.user, 1L);
                    
                ).keyBy(data -> data.f0)
                .reduce(new ReduceFunction<Tuple2<String, Long>>() 
                    @Override
                    public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception 
                        return Tuple2.of(value1.f0, value1.f1 + value2.f1);
                    
                );
        //2.选取当前最活跃的用户
        SingleOutputStreamOperator<Tuple2<String, Long>> result = clicksByUser.keyBy(data -> "key")
                .reduce(new ReduceFunction<Tuple2<String, Long>>() 
                    @Override
                    public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception 
                        return value1.f1 > value2.f1 ? value1 : value2;
                    
                );

        result.print();

        env.execute();
    

reduce 同简单聚合算子一样，也要针对每一个 key 保存状态。因为状态不会清空，所以我 们需要将 reduce 算子作用在一个有限 key 的流上。

5.3.3 用户自定义函数（UDF）

Flink 的 DataStream API 编程风格其实是一致的：基本上都 是基于 DataStream 调用一个方法，表示要做一个转换操作；方法需要传入一个参数，这个参 数都是需要实现一个接口。

很容易发现，这些接口有一个共同特点：全部都以算子操作名称 + Function 命名。

我们不仅可以通过自定义函数类或者匿名类来 实现接口，也可以直接传入 Lambda 表达式。这就是所谓的用户自定义函数（user-defined function，UDF）。

1. 函数类（Function Classes）

对于大部分操作而言，都需要传入一个用户自定义函数（UDF），实现相关操作的接口， 来完成处理逻辑的定义。Flink 暴露了所有 UDF 函数的接口，具体实现方式为接口或者抽象类， 例如 MapFunction、FilterFunction、ReduceFunction 等。

所以最简单直接的方式，就是自定义一个函数类，实现对应的接口。

下面例子实现了 FilterFunction 接口，用来筛选 url 中包含“home”的事件：

代码实现：

import org.apache.flink.api.common.functions.FilterFunction; 
import org.apache.flink.streaming.api.datastream.DataStream; 
import org.apache.flink.streaming.api.datastream.DataStreamSource; 
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; 
 
public class TransFunctionUDFTest  
 public static void main(String[] args) throws Exception  
 StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment(); 
 env.setParallelism(1); 
 

 DataStreamSource<Event> clicks = env.fromElements( 
 new Event("Mary", "./home", 1000L), 
 new Event("Bob", "./cart", 2000L) 
 ); 
 
 DataStream<Event> stream = clicks.filter(new FlinkFilter()); 
 
 stream.print(); 
 
 env.execute(); 
  
 public static class FlinkFilter implements FilterFunction<Event>  
 @Override 
 public boolean filter(Event value) throws Exception  
 return value.url.contains("home"); 
  
  
 

当然还可以通过匿名类来实现 FilterFunction 接口：

DataStream<String> stream = clicks.filter(new FilterFunction<Event>()  
 @Override 
 public boolean filter(Event value) throws Exception  
 return value.url.contains("home"); 
  
); 

为了类可以更加通用，我们还可以将用于过滤的关键字"home"抽象出来作为类的属性， 调用构造方法时传进去。

DataStream<Event> stream = clicks.filter(new KeyWordFilter("home")); 
 
public static class KeyWordFilter implements FilterFunction<Event>  
 private String keyWord; 
 
 KeyWordFilter(String keyWord)  this.keyWord = keyWord;  
 
 @Override 
 public boolean filter(Event value) throws Exception  
 return value.url.contains(this.keyWord); 
  
 

2. 匿名函数（Lambda）

匿名函数（Lambda 表达式）是 Java 8 引入的新特性，方便我们更加快速清晰地写代码。 Lambda 表达式允许以简洁的方式实现函数，以及将函数作为参数来进行传递，而不必声明额外的（匿名）类。 Flink 的所有算子都可以使用 Lambda 表达式的方式来进行编码，但是，当 Lambda 表 达式使用 Java 的泛型时，我们需要显式的声明类型信息。

下例演示了如何使用 Lambda 表达式来实现一个简单的 map() 函数，我们使用 Lambda表达式来计算输入的平方。在这里，我们不需要声明 map() 函数的输入 i 和输出参数的数据 类型，因为 Java 编译器会对它们做出类型推断。

import org.apache.flink.api.common.functions.MapFunction; 
import org.apache.flink.api.java.tuple.Tuple2; 
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; 
 
public class TransFunctionLambdaTest  
 public static void main(String[] args) throws Exception  
 StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment(); 
 env.setParallelism(1); 
 
 DataStreamSource<Event> clicks = env.fromElements( 
 new Event("Mary", "./home", 1000L), 
 new Event("Bob", "./cart", 2000L) 
 ); 
 
 //map 函数使用 Lambda 表达式，返回简单类型，不需要进行类型声明 
 DataStream<String> stream1 = clicks.map(event -> event.url); 
 
 stream1.print(); 
 
 
 env.execute(); 
  
 

由于 OUT 是 String 类型而不是泛型，所以 Flink 可以从函数签名 OUT map(IN value)

的实现中自动提取出结果的类型信息。

但是对于像 flatMap() 这样的函数，它的函数签名 void flatMap(IN value, Collector out) 被 Java 编译器编译成了 void flatMap(IN value, Collector out)，也就是说将 Collector 的泛 型信息擦除掉了。这样 Flink 就无法自动推断输出的类型信息了。

在这种情况下，我们需要显式地指定类型信息，否则输出将被视为 Object 类型，这会导 致低效的序列化。

// flatMap 使用 Lambda 表达式，必须通过 returns 明确声明返回类型 
DataStream<String> stream2 = clicks.flatMap((Event event, Collector<String> 
out) ->  
out.collect(event.url); 
).returns(Types.STRING); 
 
stream2.print();

当使用 map() 函数返回 Flink 自定义的元组类型时也会发生类似的问题。下例中的函数签 名 Tuple2 map(Event value) 被类型擦除为 Tuple2 map(Event value)。

一般来说，这个问题可以通过多种方式解决：

import org.apache.flink.api.common.functions.MapFunction; 
import org.apache.flink.api.common.typeinfo.Types; 
import org.apache.flink.api.java.tuple.Tuple2; 
import org.apache.flink.streaming.api.datastream.DataStream; 
import org.apache.flink.streaming.api.datastream.DataStreamSource; 
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; 
 
public class ReturnTypeResolve  
 public static void main(String[] args) throws Exception  
 StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment(); 
 env.setParallelism(1); 
 
 DataStreamSource<Event> clicks = env.fromElements( 
 new Event("Mary", "./home", 1000L), 
 new Event("Bob", "./cart", 2000L) 
 ); 
 
 // 想要转换成二元组类型，需要进行以下处理 
96 
 

 // 1) 使用显式的 ".returns(...)" 
 DataStream<Tuple2<String, Long>> stream3 = clicks 
 .map( event -> Tuple2.of(event.user, 1L) ) 
 .returns(Types.TUPLE(Types.STRING, Types.LONG)); 
 stream3.print(); 
 
 
 // 2) 使用类来替代 Lambda 表达式 
 clicks.map(new MyTuple2Mapper()) 
 .print(); 
 
 // 3) 使用匿名类来代替 Lambda 表达式 
 clicks.map(new MapFunction<Event, Tuple2<String, Long>>()  
 @Override 
 public Tuple2<String, Long> map(Event value) throws Exception  
 return Tuple2.of(value.user, 1L); 
  
 ).print(); 
 
 env.execute(); 
  
 
 // 自定义 MapFunction 的实现类 
 public static class MyTuple2Mapper implements MapFunction<Event, Tuple2<String, 
Long>> 
 @Override 
 public Tuple2<String, Long> map(Event value) throws Exception  
 return Tuple2.of(value.user, 1L); 
  
  
 

3. 富函数类（Rich Function Classes）

“富函数类”也是 DataStream API 提供的一个函数类的接口，所有的 Flink 函数类都有其

Rich 版本。富函数类一般是以抽象类的形式出现的。例如：RichMapFunction、RichFilterFunction、RichReduceFunction 等。

既然“富”，那么它一定会比常规的函数类提供更多、更丰富的功能。与常规函数类的不 同主要在于，富函数类可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能。

注：生命周期的概念在编程中其实非常重要，到处都有体现在 JVM 中，虚拟机会自动帮助我们管理对象的生命周期。Rich Function 有生命周期的概念。典型的生命周期方法有：

⚫ open()方法，是 Rich Function 的初始化方法，也就是会开启一个算子的生命周期。当 一个算子的实际工作方法例如 map()或者 filter()方法被调用之前，open()会首先被调 用。所以像文件 IO 的创建，数据库连接的创建，配置文件的读取等等这样一次性的 工作，都适合在 open()方法中完成。。

⚫ close()方法，是生命周期中的最后一个调用的方法，类似于解构方法。一般用来做一 些清理工作。

需要注意的是，这里的生命周期方法，对于一个并行子任务来说只会调用一次；而对应的， 实际工作方法，例如 RichMapFunction 中的 map()，在每条数据到来后都会触发一次调用。

package com.atguigu.chapter05;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class TransFromRichFunctionTest 
    public static void main(String[] args) throws Exception 
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        DataStreamSource<Event> clicks = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L),
                new Event("Alice", "./prod?id=1", 5 * 1000L),
                new Event("Cary", "./home", 60 * 1000L)
        );

        // 将点击事件转换成长整型的时间戳输出
        clicks.map(new RichMapFunction<Event, Long>() 
                    @Override
                    public void open(Configuration parameters) throws Exception 
                        super.open(parameters);
                        System.out.println(" open生命周期被调用：" +
                                getRuntimeContext().getIndexOfThisSubtask() + "号任务启动");
                    

                    @Override
                    public Long map(Event value) throws Exception 
                        return value.timestamp;

                    

                    @Override
                    public void close() throws Exception 
                        super.close();
                        System.out.println("close生命周期被调用" +
                                getRuntimeContext().getIndexOfThisSubtask() + "号任务结束");
                    
                )
                .print();

        env.execute();
    

一个常见的应用场景就是，如果我们希望连接到一个外部数据库进行读写操作，那么将连 接操作放在 map()中显然不是个好选择——因为每来一条数据就会重新连接一次数据库；所以 我们可以在 open()中建立连接，在 map()中读写数据，而在 close()中关闭连接。所以我们推荐 的最佳实践如下：

public class MyFlatMap extends RichFlatMapFunction<IN, OUT>>  
 @Override 
 public void open(Configuration configuration)  
 
 // 做一些初始化工作 
 // 例如建立一个和 mysql 的连接 
  
 
 @Override 
 public void flatMap(IN in, Collector<OUT out)  
 // 对数据库进行读写 
  
 
 @Override 
 public void close()  
 // 清理工作，关闭和 MySQL 数据库的连接。 
99 
 

 
  
 

富函数类提供了 getRuntimeContext()方法，可以获取到运行时上下文的一些信息，例如程序执行的并行度，任务名称，以及状态 （state）。这使得我们可以大大扩展程序的功能，特别是对于状态的操作，使得 Flink 中的算子 具备了处理复杂业务的能力。