DataStream（二）

旧城等待， 2023-09-27 09:19 37阅读 0赞

**目录**

5.3.2 聚合算子（Aggregation）

5.3.3 用户自定义函数（UDF）

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

**3. 扁平映射（flatMap）**

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

![fa541e801ac64b55ae88131982d76c24.png][]

同 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中进行处理了**。

![fac2ab8873b04386ba2c5079d1ea3681.png][]

在内部，是通过计算 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 中的算子 具备了处理复杂业务的能力。

[fa541e801ac64b55ae88131982d76c24.png]: https://img-blog.csdnimg.cn/fa541e801ac64b55ae88131982d76c24.png
[fac2ab8873b04386ba2c5079d1ea3681.png]: https://img-blog.csdnimg.cn/fac2ab8873b04386ba2c5079d1ea3681.png