Flink大数据分析实战

1. Flink运行架构及原理 1.1 YARN集群架构 YARN集群总体上是经典的主／从(Master/Slave)架构，主要由ResourceManager、NodeManager、ApplicationMaster和Container等几个组件构成。 一个应用程序所需的Container分为两类：运行ApplicationMaster的Container和运行各类Task的Container。前者是由ResourceManager向内部的资源调度器申请和启动的，后者是由ApplicationMaster向ResourceManager申请的，并由ApplicationMaster请求NodeManager进行启动。 总的来说，我们可以这样认为：ResourceManager管理整个集群，NodeManager管理集群中的单个节点，ApplicationMaster管理单个应用程序（集群中可能同时有多个应用程序在运行，每个应用程序都有各自的ApplicationMaster）。 1.2 Flink Standalone架构 Flink中的每一个操作算子称为一个Task（任务），例如1.7.3节的单词计数中使用的flatMap()算子、map()算子等。每个Task在一个JVM线程中执行。多个Task可以在同一个JVM进程中共享TCP连接（通过多路复用技术）和心跳信息。它们还可能共享数据集和数据结构，从而降低每个Task的开销。Task是基本的工作单元，由Flink的Runtime来执行。Task正好封装了一个算子或者算子链（见2.2.1节的任务链）的并行实例。 TaskManager为了控制执行的Task数量，将计算资源（内存）划分为多个Task Slot（任务槽），每个Task Slot代表TaskManager的一份固定内存资源，Task则在Task Slot中执行。例如，具有3个Task Slot的TaskManager会将其管理的内存资源分成3等份给每个Task Slot。这种划分方式可以更好地对多个Task进行内存隔离（没有进行CPU隔离），使其不会竞争资源。 1.3 Flink On YARN的架构 2. Flink任务调度原理 由于每个SubTask只能在一个线程中执行，为了能够减少线程间切换和缓冲的开销，在降低延迟的同时提高整体吞吐量，Flink可以将多个连续的SubTask链接成一个Task在一个线程中执行。这种将多个SubTask连在一起的方式称为任务链。 为了充分利用计算资源，提高计算效率，可以增加算子的实例数（SubTask数量）。一个特定算子的SubTask数量称为该算子的并行度，且任意两个算子的并行度之间是独立的，不同算子可能拥有不同的并行度。 4种并行度级别的作用顺序为：算子级别>执行环境级别>客户端级别>系统级别。 3. 重要概念 在Flink中，数据流或数据集被划分成多个独立的子集，这些子集分布到了不同的节点上，而每一个子集称为分区(Partition)。 DataSet和DataStream数据集都是分布式数据集，分布式数据集是指：一个数据集存储在不同的服务器节点上，每个节点存储数据集的一部分。 4. 作业流程 用户编写应用程序代码，并通过Flink客户端提交作业。程序一般为Java或Scala语言，调用Flink API构建逻辑数据流图，然后转为作业图JobGraph，并附加到StreamExecutionEnvironment中。代码和相关配置文件被编译打包，被提交到JobManager的Dispatcher，形成一个应用作业。 Dispatcher（JobManager的一个组件）接收到这个作业，启动JobManager，JobManager负责本次作业的各项协调工作。 接下来JobManager向ResourceManager申请本次作业所需的资源。 JobManager将用户作业中的作业图JobGraph转化为并行化的物理执行图，对作业并行处理并将其子任务分发部署到多个TaskManager上执行。每个作业的并行子任务将在Task Slot中执行。至此，一个Flink作业就开始执行了。 TaskManager在执行计算任务的过程中可能会与其他TaskManager交换数据，会使用相应的数据交换策略。同时，TaskManager也会将一些任务状态信息反馈给JobManager，这些信息包括任务启动、运行或终止的状态、快照的元数据等。 5. Transformation数据转换 map()算子接收一个函数作为参数，并把这个函数应用于DataStream的每个元素，最后将函数的返回结果作为结果DataStream中对应元素的值，即将DataStream的每个元素转换成新的元素。与map()算子类似，但是每个传入该函数func的DataStream元素会返回0到多个元素，最终会将返回的所有元素合并到一个DataStream。 通过函数filter对源DataStream的每个元素进行过滤，并返回一个新的DataStream。 keyBy()算子主要作用于元素类型是元组或数组的DataStream上。使用该算子可以将DataStream中的元素按照指定的key（指定的字段）进行分组，具有相同key的元素将进入同一个分区中（不进行聚合），并且不改变原来元素的数据结构。 reduce()算子主要作用于KeyedStream上，对KeyedStream数据流进行滚动聚合，即将当前元素与上一个聚合值进行合并，并且发射出新值。 connect()算子可以连接两个数据流，并保持各自元素的数据类型不变，允许在两个流之间共享状态数据。与union()算子相比，二者的区别在于：1)union()要求多个数据流的数据类型必须相同，connect()允许多个数据流中的元素类型可以不同。2)union()可以合并多个数据流，但connect()只能连接两个数据流。3)union()的执行结果是DataStream，而connect()的执行结果是ConnectedStreams；ConnectedStreams表示两个（可能）不同数据类型的连接流，可以对两个流的数据应用不同的处理方法，当一个流上的操作直接影响另一个流上的操作时，连接流非常有用。可以通过流之间的共享状态对两个流进行操作。 6. 分区策略 对于ForwardPartitioner，必须保证上下游算子并行度一致，否则会抛出异常。 全局分区策略将上游所有分区中的所有元素发送到了下游编号为0的分区中。 Key分区策略根据元素Key的Hash值输出到下游算子指定的实例。keyBy()算子底层正是使用的该分区策略，底层最终会调用KeyGroupStreamPartitioner的selectChannel()方法，计算每个Key对应的通道索引（通道编号，可理解为分区编号），根据通道索引将Key发送到下游相应的分区中。 如果想将元素均匀地输出到下游算子的每个实例，以实现负载均衡，同时又不希望使用平衡分区策略的全局负载均衡，则可以使用重新调节分区策略。该策略会尽可能避免数据在网络间传输，而能否避免还取决于TaskManager的Task Slot数量、上下游算子的并行度等。 7. 窗口计算 事件时间(Event Time)：每个事件或元素在其生产设备上产生的时间。该时间通常在它们进入Flink之前嵌入事件中，并且可以从每个事件中提取事件时间戳。流程序的时间默认使用的是事件时间。 Flink的窗口可以分为滚动窗口、滑动窗口、会话窗口、全局窗口，且每种窗口又可分别根据事件时间和处理时间进行创建。 使用滑动窗口时，需要设置窗口大小和滑动步长两个参数。滑动步长决定了Flink以多大的频率来创建新的窗口，步长较小，窗口的个数会很多。步长小于窗口的大小时，相邻窗口会重叠，一个事件会被分配到多个窗口；步长大于窗口大小时，有些事件可能会丢失。 ReduceFunction指定如何聚合输入中的两个元素以产生相同类型的输出元素。 AggregateFunction是聚合函数的基本接口，也是ReduceFunction的通用版本。与ReduceFunction相同，Flink将在窗口输入元素到达时对其进行增量聚合。 使用ProcessWindowFunction可以获得一个包含窗口所有元素的可迭代对象(Iterable)，以及一个可以访问时间和状态信息的上下文对象(Context)，这使得它比其他窗口函数提供了更多的灵活性。 由于ProcessWindowFunction是全量计算函数，如果既要获得窗口信息又要进行增量聚合，则可以将ProcessWindowFunction与ReduceFunction或AggregateFunction结合使用。 Flink不保证窗口内元素的顺序。这意味着，虽然清除器可以从窗口的开头移除元素，但这些元素不一定是最先到达的。 8. 水印 为了保证计算结果的正确性，需要让窗口等待延迟数据到达后再进行计算，但是不能无限期地等待下去，必须有一种机制来确定何时触发窗口计算，这种机制就是水印(Watermark)。 水印=进入Flink的当前最大事件时间‒允许最大延迟时间，当水印>=窗口结束时间时，立即触发窗口计算，计算完毕后发射出计算结果并销毁窗口；否则窗口将一直等待。 对于一些使用水印解决不了的、更为严重的延迟数据，Flink默认是丢弃的，为了保证数据不丢失，Flink提供了允许延迟(Allowed Lateness)和侧道输出机制(Side Output)。 由于下游算子对水印的计算方式是取所有不同的上游并行算子中水印的最小值，如果上游某一个算子因为缺少数据迟迟没有生成水印，就会出现事件时间倾斜问题，水印将不会发生变化，导致下游不能触发计算。 为了解决这个问题，可以使用WatermarkStrategy接口中已经实现的withIdleness(Duration idleTimeout)方法来检测空闲输入并将其标记为空闲状态。 9. 状态管理 9.1 总结来说，Flink状态管理的主要特性如下： 本地性：Flink状态是存储在使用它的机器本地的，并且可以内存访问速度来获取。 持久性：Flink状态是容错的，例如它可以自动按一定的时间间隔产生快照，并且在任务失败后进行恢复。· 纵向可扩展性：Flink状态可以存储在集成的RocksDB（一种KV型数据库）实例中，这种方式下可以通过增加本地磁盘来扩展空间。· 横向可扩展性：Flink状态可以随着集群的扩／缩容重新分布。· 可查询性：Flink状态可以通过使用状态查询API从外部进行查询。 9.2 AggregatingState<IN, OUT>：保留一个单值。和ReducingState不同的是，聚合类型可能与添加到状态的元素的类型不同。使用add(IN)添加元素，并与已有的元素使用AggregateFunction进行聚合。 10. 容错机制◆ Flink将Checkpoint快照的存储位置称为状态后端(State Backend)。状态后端有两种实现：一种基于RocksDB将工作状态保存在磁盘上，使用RocksDBStateBackend类型；另一种基于堆将工作状态保存在Java的堆内存中。基于堆的状态后端有两种类型：FsStateBackend和MemoryStateBackend。FsStateBackend会定时将其状态快照持久化到分布式文件系统中，MemoryStateBackend使用JobManager的堆内存保存状态快照。 ◆ RocksDBStateBackend适用于状态非常大、窗口非常长的作业以及所有高可用的场景。由于其允许存储非常大的状态，因此意味着使用RocksDBStateBackend将会使应用程序的最大吞吐量降低。所有的读写都必须序列化、反序列化操作，比基于堆内存的状态后端的效率要低很多。 ◆ FsStateBackend状态后端适用于状态比较大、窗口比较长的作业以及所有高可用的场景。对于一些以分钟为单位的窗口聚合，建议使用该状态后端。 ◆ 通常，建议在生产中避免使用MemoryStateBackend，因为它将快照存储在JobManager的内存中，而不是持久化到磁盘。当需要在FsStateBackend和RocksDBStateBackend之间进行选择时，需要从性能和可伸缩性方面进行考虑。FsStateBackend非常快，因为每个状态访问和更新操作在Java堆内存上，但是状态大小受集群内可用内存的限制。另一方面，RocksDBStateBackend可以根据可用磁盘空间进行扩展，并且是唯一支持增量快照的状态后端。但是每个状态访问和更新都需要序列化／反序列化，这样会导致平均性能比内存状态后端慢一个数量级。 ◆ Savepoint（保存点）是用户手动触发的Checkpoint，由用户手动创建、拥有和删除，它获取状态的快照并将其写入状态后端。Savepoint主要用于手动的状态数据备份和恢复，常用于在升级和维护集群的过程中保存状态数据，避免系统无法恢复到原有的计算状态。例如，需要升级程序之前先执行一次Savepoint，升级后可以继续从升级前的那个点进行计算，保证数据不中断。 11. SQL API TUMBLE函数将每个元素分配给一个具有指定窗口大小的滚动窗口。滚动窗口有一个固定的大小，不重叠。 ◆ HOP窗口函数将元素分配给固定长度的窗口。与TUMBLE窗口函数一样，窗口的大小是由窗口大小参数配置的。另一个滑动步伐参数控制HOP窗口的启动频率（即间隔多长时间开启一个新窗口）。因此，如果滑动步伐小于窗口大小，HOP窗口可以重叠。HOP窗口也被称为“滑动窗口”。 ◆ CUMULATE窗口在某些情况下是非常有用的，例如在固定的窗口间隔内触发滚动窗口。 ◆ 窗口聚合定义在GROUP BY子句中，该子句包含应用于窗口函数的window_start和window_end列。与使用常规GROUP BY子句的查询一样，使用按窗口分组聚合的查询将为每个组计算单个结果行。 ◆ 窗口聚合也支持GROUPING SETS语法。该语法允许比标准GROUP BY所描述的更复杂的分组操作。其可以对GROUP BY分组后的每一组继续按照指定的列进行分组，并为每个组计算聚合，就像简单的GROUP BY子句一样。 ◆ ROLLUP是一种用于指定通用类型分组集的简写符号。它表示给定的表达式列表和列表的所有前缀，包括空列表。使用ROLLUP的窗口聚合要求window_start和window_end列都必须在GROUP BY子句中，而不是在ROLLUP子句中。 ◆ CUBE是一种用于指定通用类型分组集的简写符号，它表示给定的列表及其所有可能的子集。 ◆ OVER聚合计算一段有序行范围内的每个输入行的聚合值。 12. 流图&执行图 1)Flink Client将作业的应用程序代码生成StreamGraph（在批模式下生成的是OptimizedPlan，这里暂不讲解）。StreamGraph是表示流处理程序拓扑的数据结构，描述了算子和算子之间逻辑上的拓扑关系，封装了生成作业图(JobGraph)的必要信息。 2)将StreamGraph转换为JobGraph。JobGraph表示JobManager可接受的低级别Flink数据流程序。所有来自更高级别API的程序都被转换成JobGraph。JobGraph是一个顶点和中间结果的图，这些顶点和中间结果连接在一起形成一个DAG（有向无环图）。JobGraph定义了作业范围的配置信息，而每个顶点和中间结果定义了具体操作和中间数据的特征。3)将JobGraph提交给Dispatcher。Dispatcher组件负责接收作业提交、在故障时恢复作业、监控Flink会话集群的状态等。4)Dispatcher根据JobGraph创建相应的JobManager并运行。5)JobManager将JobGraph转换为ExecutionGraph。ExecutionGraph是协调数据流的分布式执行的中心数据结构，它保留了每个并行任务、每个中间流以及它们之间的通信信息。6)JobManager将ExecutionGraph转换为物理执行图。 ◆ StreamNode和StreamEdge之间有着双向的依赖关系。StreamEdge包含其连接的源节点（源StreamNode，使用sourceVertex属性表示）和目的节点（目的StreamNode，使用sourceVertex属性表示） ◆ StreamGraph和JobGraph的转化生成都是在Flink客户端，而最终Flink作业运行时调度层的核心执行图ExecutionGraph是在服务端的JobManager中生成的。 ◆ 总的来说，ExecutionGraph的转换过程为：将JobGraph按照拓扑排序后得到一个JobVertex集合，遍历该JobVertex集合，即从Source开始，将JobVertex封装成ExecutionJobVertex，并依次创建ExecutionVertex、Execution、IntermediateResult和IntermediateResultPartition。然后通过ejv.connectToPredecessor()方法创建ExecutionEdge，建立当前节点与其上游节点之间的联系，即连接ExecutionVertex和IntermediateResultPartition。构建好ExecutionGraph后，接下来会基于ExecutionGraph触发作业的调度，申请Task Slot，部署任务到TaskManager执行。

我是本书作者，作者微信jock90，可与我交流疑难问题，也可加入读者群相互学习。

推荐给初学者，常用使用场景都覆盖了。