17.Pojo类

Flink可以从各种来源获取数据，然后构建DataStream进行转换处理。一般将数据的输入来源称为数据源（data source），而读取数据的算子就是源算子（source operator）。所以，source就是我们整个处理程序的输入端。

在Flink1.12以前，旧的添加source的方式，是调用执行环境的addSource()方法：

DataStream<String> stream = env.addSource(...);

方法传入的参数是一个“源函数”（source function），需要实现SourceFunction接口。

从Flink1.12开始，主要使用流批统一的新Source架构：

DataStreamSource<String> stream = env.fromSource(…)

Flink直接提供了很多预实现的接口，此外还有很多外部连接工具也帮我们实现了对应的Source，通常情况下足以应对我们的实际需求。

5.2.1 准备工作

为了方便练习，这里使用WaterSensor作为数据模型。

字段名	数据类型	说明
id	String	水位传感器类型
ts	Long	传感器记录时间戳
vc	Integer	水位记录

具体代码如下：

public class WaterSensor {
    public String id;
    public Long ts;
    public Integer vc;

    public WaterSensor() {
    }

    public WaterSensor(String id, Long ts, Integer vc) {
        this.id = id;
        this.ts = ts;
        this.vc = vc;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }

    public Long getTs() {
        return ts;
    }

    public void setTs(Long ts) {
        this.ts = ts;
    }

    public Integer getVc() {
        return vc;
    }

    public void setVc(Integer vc) {
        this.vc = vc;
    }

    @Override
    public String toString() {
        return "WaterSensor{" +
                "id='" + id + '\'' +
                ", ts=" + ts +
                ", vc=" + vc +
                '}';
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        WaterSensor that = (WaterSensor) o;
        return Objects.equals(id, that.id) &&
                Objects.equals(ts, that.ts) &&
                Objects.equals(vc, that.vc);
    }

    @Override
    public int hashCode() {

        return Objects.hash(id, ts, vc);
    }
}

这里需要注意，我们定义的WaterSensor，有这样几个特点：

1. 类是公有（public）的
2. 有一个无参的构造方法
3. 所有属性都是公有（public）的
4. 所有属性的类型都是可以序列化的

Flink会把这样的类作为一种特殊的POJO（Plain Ordinary Java Object简单的Java对象，实际就是普通JavaBeans）数据类型来对待，方便数据的解析和序列化。另外我们在类中还重写了toString方法，主要是为了测试输出显示更清晰。

我们这里自定义的POJO类会在后面的代码中频繁使用，所以在后面的代码中碰到，把这里的POJO类导入就好了。

18.Lambda表达式

源算子（source）

Flink 可以从各种来源获取数据，然后构建 DataStream 进行转换处理。一般将数据的输入来源称为数据源(data source)，而读取数据的算子就是源算子（source operator）。所以，source就是我们整个处理程序的输入端。

Flink 代码中通用的添加 source 的方式，是调用执行环境的 addSource()方法：

DataStream<String> stream = env.addSource(...);

方法传入一个对象参数，需要实现 SourceFunction 接口；返回 DataStreamSource。这里的DataStreamSource 类继承自 SingleOutputStreamOperator 类，又进一步继承自 DataStream。所以很明显，读取数据的 source 操作是一个算子，得到的是一个数据流（DataStream）。

源算子（source）：准备工作

为了更好地理解，我们先构建一个实际应用场景。比如网站的访问操作，可以抽象成一个三元组（用户名，用户访问的 url，用户访问 url 的时间戳），所以在这里，我们可以创建一个类 Event，将用户行为包装成它的一个对象。

Event类字段设计：

包下创建一个Event类

package com.atguigu.chapter05;
import java.sql.Timestamp;
public class Event {
    public String user;
    public String url;
    public Long timestamp;
    public Event() {
    }
    public Event(String user, String url, Long timestamp) {
        this.user = user;
        this.url = url;
        this.timestamp = timestamp;
    }
    @Override
    public String toString() {
        return "Event{" +
                "user='" + user + '\'' +
                ", url='" + url + '\'' +
                ", timestamp=" + new Timestamp(timestamp) +
                '}';
    }
}

基本转换算子(上篇已有提及跳转链接）

1. 映射（map）

结果输出：

result1> Mary
result1> Bob
result2> Mary
result2> Bob
result3> Mary
result3> Bob

lambda表达式

SingleOutputStreamOperator<String> result3 = stream.map(data -> data.user);

1. data -> data.user：

   • data 是传入 map 的每个 Event 对象（输入类型）。
   • data.user 是从每个 Event 对象中提取 user 字段（输出类型）。

   这里的 Lambda 表达式等价于这样一个匿名类：

   SingleOutputStreamOperator<String> result3 = stream.map(new MapFunction<Event, String>() {
       @Override
       public String map(Event data) throws Exception {
           return data.user;
       }
   });
   

2. SingleOutputStreamOperator<String>：该类型表示 map 操作后的输出流，这里是 String 类型的流（因为我们将 Event 对象转换为了 user 字段，即字符串类型）。

2. 过滤（filter）

结果输出：

result1>Event{user='Mary', url='./home', timestamp=1970-01-01 08:00:01.0}
result2>Event{user='Bob', url='./cart', timestamp=1970-01-01 08:00:02.0}
result3>Event{user='Bob', url='./cart', timestamp=1970-01-01 08:00:02.0}

lambda表达式

SingleOutputStreamOperator<Event> result3 = stream.filter(data -> data.user.equals("Bob"));

data -> data.user.equals("Bob") 是一个 Lambda 表达式，表示对 Event 对象进行过滤，保留 user 字段为 "Bob" 的事件。

3. 扁平映射（flatMap））

方法一使用自定义 FlatMapFunction，对每个 Event 输出三个字段：

• 对于 Mary：
  • "Mary"
  • "./home"
  • "1000"
• 对于 Bob：
  • "Bob"
  • "./cart"
  • "2000"

输出（标签为 "1"）：

1> Mary
1> ./home
1> 1000
1> Bob
1> ./cart
1> 2000

方法二使用 Lambda 表达式，转换逻辑如下：

• 对于 Mary：
  • user 不是 "Marry"（注意拼写不同），所以不满足第一个 if 条件。
  • 不满足第二个 else if 条件（user 不是 "Bob"）。
  • 收集 url 和 timestamp。
  • 输出：
    • "./home"
    • "1000"
• 对于 Bob：
  • 不满足第一个 if 条件。
  • 满足第二个 else if 条件。
  • 收集 user。
  • 收集 url 和 timestamp。
  • 输出：
    • "Bob"
    • "./cart"
    • "2000"

输出（标签为 "2"）：

2> ./home
2> 1000
2> Bob
2> ./cart
2> 2000

19.富函数类生命周期

富函数类（Rich Function Classes）

“富函数类”也是 DataStream API 提供的一个函数类的接口，所有的 Flink 函数类都有其Rich 版本。富函数类一般是以抽象类的形式出现的。

与常规函数类的不同主要在于，富函数类可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能。

Rich Function 有生命周期的概念。典型的生命周期方法有：

20.时间语义

在事件发生之后，生成的数据被收集起来，首先进入分布式消息队列，然后被 Flink 系统中的 Source 算子读取消费，进而向下游的转换算子（窗口算子）传递，最终由窗口算子进行计算处理。

这里有两个非常重要的时间点：一个是数据产生的时间，我们把它叫作“事件时间”（Event Time）；另一个是数据真正被处理的时刻，叫作“处理时间”（Processing Time）。我们所定义的窗口操作，到底是以那种时间作为衡量标准，就是所谓的“时间语义”（Notions of Time）。由于分布式系统中网络传输的延迟和时钟漂移，处理时间相对事件发生的时间会有所滞后。

1. 处理时间（Processing Time）

处理时间的概念非常简单，就是指执行处理操作的机器的系统时间。

如果我们以它作为衡量标准，那么数据属于哪个窗口就很明显了：只看窗口任务处理这条数据时，当前的系统时间。比如之前举的例子，数据 8 点 59 分 59 秒产生，而窗口计算时的时间是 9 点零 1 秒，那么这条数据就属于 9 点—10 点的窗口；如果数据传输非常快，9 点之前就到了窗口任务，那么它就属于 8 点—9 点的窗口了。每个并行的窗口子任务，就只按照自己的系统时钟划分窗口。假如我们在早上 8 点 10 分启动运行程序，那么接下来一直到 9 点以前处理的所有数据，都属于第一个窗口；9 点之后、10 点之前的所有数据就将属于第二个窗口。这种方法非常简单粗暴，不需要各个节点之间进行协调同步，也不需要考虑数据在流中的位置，简单来说就是“我的地盘听我的”。所以处理时间是最简单的时间语义。

2. 事件时间（Event Time）

事件时间，是指每个事件在对应的设备上发生的时间，也就是数据生成的时间。

数据一旦产生，这个时间自然就确定了，所以它可以作为一个属性嵌入到数据中。这其实就是这条数据记录“时间戳”（Timestamp）。在事件时间语义下，我们对于时间的衡量，就不看任何机器的系统时间了，而是依赖于数据本身。打个比方，这相当于任务处理的时候自己本身是没有时钟的，所以只好来一个数据就问一下“现在几点了”；而数据本身也没有表，只有一个自带的“出厂时间”，于是任务就基于这个时间来确定自己的时钟。由于流处理中数据是源源不断产生的，一般来说，先产生的数据也会先被处理，所以当任务不停地接到数据时，它们的时间戳也基本上是不断增长的，就可以代表时间的推进。

哪种时间语义更重要

1）从《星球大战》说起

为了更加清晰地说明两种语义的区别，我们来举一个非常经典的例子：电影《星球大战》。

如上图所示，我们会发现，看电影其实就是处理影片中数据的过程，所以影片的上映时间就相当于“处理时间”；而影片的数据就是所描述的故事，它所发生的背景时间就相当于“事件时间”。两种时间语义都有各自的用途，适用于不同的场景。

2）数据处理系统中的时间语义

在实际应用中，事件时间语义会更为常见。一般情况下，业务日志数据中都会记录数据生成的时间戳（timestamp），它就可以作为事件时间的判断基础。

在Flink中，由于处理时间比较简单，早期版本默认的时间语义是处理时间；而考虑到事件时间在实际应用中更为广泛，从Flink1.12版本开始，Flink已经将事件时间作为默认的时间语义了。

21.水位线特性

水位线就代表了当前的事件时间时钟，而且可以在数据的时间戳基础上加一些延迟来保证不丢数据，这一点对于乱序流的正确处理非常重要。

水位线的特性：

22.Yarn模式部署

YARN上部署的过程是：客户端把Flink应用提交给Yarn的ResourceManager，Yarn的ResourceManager会向Yarn的NodeManager申请容器。在这些容器上，Flink会部署JobManager和TaskManager的实例，从而启动集群。Flink会根据运行在JobManager上的作业所需要的Slot数量动态分配TaskManager资源。

3.5.1 相关准备和配置

在将Flink任务部署至YARN集群之前，需要确认集群是否安装有Hadoop，保证Hadoop版本至少在2.2以上，并且集群中安装有HDFS服务。

具体配置步骤如下：

（1）配置环境变量，增加环境变量配置如下：

$ sudo vim /etc/profile.d/my_env.sh

HADOOP_HOME=/opt/module/hadoop-3.3.4

export PATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin

export HADOOP_CONF_DIR=${HADOOP_HOME}/etc/hadoop

export HADOOP_CLASSPATH=`hadoop classpath`

（2）启动Hadoop集群，包括HDFS和YARN。

[atguigu@hadoop102 hadoop-3.3.4]$ start-dfs.sh

[atguigu@hadoop103 hadoop-3.3.4]$ start-yarn.sh

（3）在hadoop102中执行以下命令启动netcat。

[atguigu@hadoop102 flink-1.17.0]$ nc -lk 7777

3.5.2 会话模式部署

YARN的会话模式与独立集群略有不同，需要首先申请一个YARN会话（YARN Session）来启动Flink集群。具体步骤如下：

1）启动集群

（1）启动Hadoop集群（HDFS、YARN）。

（2）执行脚本命令向YARN集群申请资源，开启一个YARN会话，启动Flink集群。

[atguigu@hadoop102 flink-1.17.0]$ bin/yarn-session.sh -nm test

可用参数解读：

1. -d：分离模式，如果你不想让Flink YARN客户端一直前台运行，可以使用这个参数，即使关掉当前对话窗口，YARN session也可以后台运行。
2. -jm（--jobManagerMemory）：配置JobManager所需内存，默认单位MB。
3. -nm（--name）：配置在YARN UI界面上显示的任务名。
4. -qu（--queue）：指定YARN队列名。
5. -tm（--taskManager）：配置每个TaskManager所使用内存。

注意：Flink1.11.0版本不再使用-n参数和-s参数分别指定TaskManager数量和slot数量，YARN会按照需求动态分配TaskManager和slot。所以从这个意义上讲，YARN的会话模式也不会把集群资源固定，同样是动态分配的。

YARN Session启动之后会给出一个Web UI地址以及一个YARN application ID，如下所示，用户可以通过Web UI或者命令行两种方式提交作业。

2022-11-17 15:20:52,711 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - Found Web Interface hadoop104:40825 of application 'application_1668668287070_0005'.

JobManager Web Interface: http://hadoop104:40825

2）提交作业

（1）通过Web UI提交作业

这种方式比较简单，与上文所述Standalone部署模式基本相同。

（2）通过命令行提交作业

① 将FlinkTutorial-1.0-SNAPSHOT.jar任务上传至集群。

② 执行以下命令将该任务提交到已经开启的Yarn-Session中运行。

[atguigu@hadoop102 flink-1.17.0]$ bin/flink run

-c com.atguigu.wc.SocketStreamWordCount FlinkTutorial-1.0-SNAPSHOT.jar

客户端可以自行确定JobManager的地址，也可以通过-m或者-jobmanager参数指定JobManager的地址，JobManager的地址在YARN Session的启动页面中可以找到。

③ 任务提交成功后，可在YARN的Web UI界面查看运行情况。hadoop103:8088。

从上图中可以看到我们创建的Yarn-Session实际上是一个Yarn的Application，并且有唯一的Application ID。

④也可以通过Flink的Web UI页面查看提交任务的运行情况，如下图所示。

23.作业提交命令

具体步骤如下：

（1） 一般情况下，由客户端（App）通过分发器提供的 REST 接口,将作业提交给JobManager。

（2）由分发器启动 JobMaster，并将作业（包含 JobGraph）提交给 JobMaster。

（3）JobMaster 将 JobGraph 解析为可执行的 ExecutionGraph，得到所需的资源数量，然后向资源管理器请求资源（slots）。

（4）资源管理器判断当前是否由足够的可用资源；如果没有，启动新的 TaskManager。

（5）TaskManager 启动之后，向 ResourceManager 注册自己的可用任务槽（slots）。

（6）资源管理器通知 TaskManager 为新的作业提供 slots。

（7）TaskManager 连接到对应的 JobMaster，提供 slots。

（8）JobMaster 将需要执行的任务分发给 TaskManager。

（9）TaskManager 执行任务，互相之间可以交换数据。

如果部署模式不同，或者集群环境不同（例如 Standalone、YARN、K8S 等），其中一些步骤可能会不同或被省略，也可能有些组件会运行在同一个 JVM 进程中。

1. Standalone模式作业提交流程

在独立模式（Standalone）下，只有会话模式和应用模式两种部署方式。两者整体来看流程是非常相似的：TaskManager 都需要手动启动，所以当 ResourceManager 收到 JobMaster 的请求时，会直接要求 TaskManager 提供资源。而 JobMaster 的启动时间点，会话模式是预先启动，应用模式则是在作业提交时启动。

2.YARN会话模式作业提交流程

提交作业的流程步骤如下：

（1）客户端通过 REST 接口，将作业提交给分发器。

（2）分发器启动 JobMaster，并将作业（包含 JobGraph）提交给 JobMaster。

（3）JobMaster 向资源管理器请求资源（slots）。

（4）资源管理器向 YARN 的资源管理器请求 container 资源。

3.YARN单作业模式任务提交流程

提交作业的流程步骤如下：

（1）客户端将作业提交给 YARN 的资源管理器，这一步中会同时将 Flink 的 Jar 包和配置上传到 HDFS，以便后续启动 Flink 相关组件的容器。

（2）YARN 的资源管理器分配 Container 资源，启动 Flink JobManager，并将作业提交给JobMaster。这里省略了 Dispatcher 组件。

（3）JobMaster 向资源管理器请求资源（slots）。

（4）资源管理器向 YARN 的资源管理器请求 container 资源。

（5）YARN 启动新的 TaskManager 容器。

（6）TaskManager 启动之后，向 Flink 的资源管理器注册自己的可用任务槽。

（7）资源管理器通知 TaskManager 为新的作业提供 slots。

（8）TaskManager 连接到对应的 JobMaster，提供 slots。

（9）JobMaster 将需要执行的任务分发给 TaskManager，执行任务。

24.窗口的概念、分类

概念

Flink 是一种流式计算引擎，主要是来处理无界数据流的，数据源源不断、无穷无尽。想要更加方便高效地处理无界流，一种方式就是将无限数据切割成有限的“数据块”进行处理，这就是所谓的“窗口”（Window）。

如果我们采用事件时间语义，就会有些费解了。由于有乱序数据，我们需要设置一个延迟时间来等所有数据到齐。比如上面的例子中，我们可以设置延迟时间为 2 秒，如图所示，这样 0~10 秒的窗口会在时间戳为 12 的数据到来之后，才真正关闭计算输出结果，这样就可以正常包含迟到的 9 秒数据了。

但是这样一来，0~10 秒的窗口不光包含了迟到的 9 秒数据，连 11 秒和 12 秒的数据也包含进去了。我们为了正确处理迟到数据，结果把早到的数据划分到了错误的窗口——最终结果都是错误的。

在 Flink 中，窗口其实并不是一个“框”，流进来的数据被框住了就只能进这一个窗口。相比之下，我们应该把窗口理解成一个“桶”，如图所示。在 Flink 中，窗口可以把流切割成有限大小的多个“存储桶”（bucket)；每个数据都会分发到对应的桶中，当到达窗口结束时间时，就对每个桶中收集的数据进行计算处理。

分类

1.按照驱动类型分类

（1）时间窗口（Time Window）

（2）计数窗口（Count Window）

窗口本身是截取有界数据的一种方式，所以窗口一个非常重要的信息其实就是“怎样截取数据”。换句话说，就是以什么标准来开始和结束数据的截取，我们把它叫作窗口的“驱动类型”。

我们最容易想到的就是按照时间段去截取数据，这种窗口就叫作“时间窗口”（Time Window）。这在实际应用中最常见，之前所举的例子也都是时间窗口。除了由时间驱动之外，窗口其实也可以由数据驱动，也就是说按照固定的个数，来截取一段数据集，这种窗口叫作“计数窗口”（Count Window）。

2.按照窗口分配数据的规则分类

（1）滚动窗口（Tumbling Windows）

滚动窗口有固定的大小，是一种对数据进行“均匀切片”的划分方式。窗口之间没有重叠，也不会有间隔，是“首尾相接”的状态。如果我们把多个窗口的创建，看作一个窗口的运动，那就好像它在不停地向前“翻滚”一样。这是最简单的窗口形式，我们之前所举的例子都是滚动窗口。也正是因为滚动窗口是“无缝衔接”，所以每个数据都会被分配到一个窗口，而且只会属于一个窗口。

（2）滑动窗口（Sliding Windows）

既然是向前滑动，那么每一步滑多远，就也是可以控制的。所以定义滑动窗口的参数有两个：除去窗口大小（window size）之外，还有一个“滑动步长”（window slide），它其实就代表了窗口计算的频率。滑动的距离代表了下个窗口开始的时间间隔，而窗口大小是固定的，所以也就是两个窗口结束时间的间隔；窗口在结束时间触发计算输出结果，那么滑动步长就代表了计算频率。

（3）会话窗口（Session Windows）

会话窗口顾名思义，是基于“会话”（session）来来对数据进行分组的。这里的会话类似Web 应用中 session 的概念，不过并不表示两端的通讯过程，而是借用会话超时失效的机制来描述窗口。简单来说，就是数据来了之后就开启一个会话窗口，如果接下来还有数据陆续到来，那么就一直保持会话；如果一段时间一直没收到数据，那就认为会话超时失效，窗口自动关闭。这就好像我们打电话一样，如果时不时总能说点什么，那说明还没聊完；如果陷入了尴尬的沉默，半天都没话说，那自然就可以挂电话了。

（4）全局窗口（Global Windows）

这种窗口全局有效，会把相同 key 的所有数据都分配到同一个窗口中；说直白一点，就跟没分窗口一样。无界流的数据永无止尽，所以这种窗口也没有结束的时候，默认是不会做触发计算的。如果希望它能对数据进行计算处理，还需要自定义“触发器”（Trigger）