Flink

​

1 Flink跟Spark Streaming的区别?

  1）Flink 是标准的实时处理引擎，基于事件驱动。而 Spark Streaming 是微批（Micro-Batch）的模型;

  2）架构模型

  Spark Streaming 在运行时的主要角色包括：Master、Worker、Driver、Executor，Flink 在运行时主要包含：Jobmanager、Taskmanager和Slot。

  3）任务调度

  Spark Streaming 连续不断的生成微小的数据批次，构建有向无环图DAG，Spark Streaming 会依次创建 DStreamGraph、JobGenerator、JobScheduler。Flink 根据用户提交的代码生成 StreamGraph，经过优化生成 JobGraph，然后提交给 JobManager进行处理，JobManager 会根据 JobGraph 生成 ExecutionGraph，ExecutionGraph 是 Flink 调度最核心的数据结构，JobManager 根据 ExecutionGraph 对 Job 进行调度。

  4）时间机制

  Spark Streaming 支持的时间机制有限，只支持处理时间。 Flink 支持了流处理程序在时间上的三个定义：处理时间、事件时间、注入时间。同时也支持 watermark 机制来处理滞后数据。

  5）容错机制

  对于 Spark Streaming 任务，我们可以设置 checkpoint，然后假如发生故障并重启，我们可以从上次 checkpoint 之处恢复，但是这个行为只能使得数据不丢失，可能会重复处理，不能做到恰好一次处理语义。Flink 则使用两阶段提交协议来解决这个问题。

2 Flink 集群运行时角色

  Flink程序在运行时主要有TaskManager，JobManager，Client三种角色;

  Client 不是运行时和程序执行的一部分， 而是用于准备数据流并将其发送给JobManager。 之后， 客户端可以断开连接（ 分离模式） ， 或保持连接来接收进程报告（ 附加模式） 。 客户端可以作为触发执行 Java/Scala 程序的一部分运行，也可以在命令行进程 ./bin/flink run … 中运行。

  可以通过多种方式启动 JobManager 和 TaskManager： 直接在机器上作为standalone 集群启动、 在容器中启动、 或者通过 YARN 等资源框架管理并启动。TaskManager 连接到 JobManagers， 宣布自己可用， 并被分配工作。

2.1 J o b M a n a g e r

  JobManager 具有许多与协调 Flink 应用程序的分布式执行有关的职责： 它决定何时调度下一个 task（ 或一组 task） 、 对完成的 task 或执行失败做出反应、协调 checkpoint、 并且协调从失败中恢复等等。 这个进程由三个不同的组件组成：

ResourceManager

  ResourceManager 负责 Flink 集群中的资源提供、 回收、 分配， 管理 task slots。

Dispatcher

  Dispatcher 提供了一个 REST 接口， 用来提交 Flink 应用程序执行， 并为每个提交的作业启动一个新的 JobMaster。 它还运行 Flink WebUI 用来提供作业执行信息。

JobMaster

  JobMaster 负责管理单个 JobGraph 的执行。 Flink 集群中可以同时运行多个作业， 每个作业都有自己的 JobMaster。

2.2 T a s k M a n a g e r

  TaskManager（ 也称为 worker） 执行作业流的 task， 并且缓存和交换数据流。

  必须始终至少有一个 TaskManager。 在 TaskManager 中资源调度的最小单位是task slot。 TaskManager 中 task slot 的数量表示并发处理 task 的数量。 请注意一个 task slot 中可以执行多个算子。

3 并行度

  Flink中的任务被分为多个并行任务来执行，其中每个并行的实例处理一部分数据。这些并行实例的数量被称为并行度。我们在实际生产环境中可以从四个不同层面设置并行度：

    ① 操作算子层面(Operator Level)

    ② 执行环境层面(Execution Environment Level)

    ③ 客户端层面(Client Level)

    ④ 系统层面(System Level)

优先级：算子层面>环境层面>客户端层面>系统层面

  并行度的设置：一般设为kafka的分区数，达到1：1

         遵循2的n次方：比如2、4、8、16……

4 Flink的keyby怎么实现的分区？分区、分组的区别？

  Keyby实现原理：

    对指定的key调用自身的hashCode方法=》hash1

    调用murmruhash算法，进行第二次hash =》键组ID

    通过一个公式，计算出当前数据应该去往哪个下游分区：

键组id * 下游算子并行度 / 最大并行度（默认128）

  分区：算子的一个并行实例可以理解成一个分区，是物理上的资源

  分组：数据根据key进行区分，是一个逻辑上的划分

  一个分区可以有多个分组，同一个分组的数据肯定在同一个分区

5 状态编程、状态机制(Flink 状态(State)管理与恢复)

  算子状态：作用范围是算子，算子的多个并行实例各自维护一个状态

  键控状态：每个分组维护一个状态

  状态后端：两件事=》 本地状态存哪里、checkpoint存哪里

6 Flink Time、Watermark、Window

6.1 Flink的三种时间语义

  Event Time：是事件创建的时间。它通常由事件中的时间戳描述，例如采集的日志数据中，每一条日志都会记录自己的生成时间，Flink通过时间戳分配器访问事件时间戳。

  Ingestion Time：是数据进入Flink的时间。

  Processing Time：是每一个执行基于时间操作的算子的本地系统时间，与机器相关，默认的时间属性就是Processing Time。

6.2 Flink 中的Watermark机制

  1）Watermark 是一种衡量 Event Time 进展的机制，可以设定延迟触发；

  2）Watermark 是用于处理乱序事件的，而正确的处理乱序事件，通常用Watermark 机制结合 window 来实现；

  3）基于事件时间，用来触发窗口、定时器等；

  4）watermark主要属性就是时间戳，可以理解一个特殊的数据，插入到流里面；

  5）watermark是单调不减的；

  6）数据流中的 Watermark 用于表示 timestamp 小于 Watermark 的数据，都已经到达了，如果后续还有timestamp 小于 Watermark 的数据到达，称为迟到数据。

6.3 Watermark是数据吗？怎么生成的？怎么传递的？

  Watermark是一条携带时间戳的特殊数据，从代码指定生成的位置，插入到流里面。

  一对多：广播

  多对一：取最小

  多对多：拆分来看，其实就是上面两种的结合

6.4 Watermark的生成方式？

  间歇性：来一条数据，更新一次watermark

  周期性：固定周期更新watermark

  官方提供的api是基于周期的，默认200ms，因为间歇性会给系统带来压力。

Watermark=当前最大事件时间-乱序时间-1ms

6.5 窗口（分类、生命周期、触发、划分）

6.5.1 窗口分类： Keyed Window和Non-keyed Window

  基于时间：滚动、滑动、会话

  基于数量：滚动、滑动

6.5.2 Window口的4个相关重要组件：

  assigner（分配器）：如何将元素分配给窗口

  function（计算函数）：为窗口定义的计算。其实是一个计算函数，完成窗口内容的计算

  triger（触发器）：在什么条件下触发窗口的计算

  evictor（退出器）：定义从窗口中移除数据

6.5.3 窗口的划分

  start=按照数据的事件时间向下取窗口长度的整数倍

  end=start+size

    比如开了一个10s的滚动窗口，第一条数据是857s，那么它属于[850s,860s)

6.5.4 窗口的创建

  当属于某个窗口的第一个元素到达，Flink就会创建一个窗口

6.5.5 窗口的销毁

  当时间超过其结束时间+用户指定的允许延迟时间（Flink保证只删除基于时间的窗口，而不能删除其他类型的窗口，例如全局窗口）。

6.5.6 窗口为什么左闭右开

  属于窗口的最大时间戳=end-1ms

6.5.7 窗口什么时候触发

  如基于事件时间的窗口 watermark>=end-1ms

7 Exactly-Once的保证

  一般说的是端到端一致性，要考虑source和sink：

    Source：可重发

    Flink内部：Checkpoint机制（介绍Chandy-Lamport算法、barrier对齐）

    Sink：幂等性 或 事务性 写入

  我们使用的Source和Sink主要是Kafka：

    作为source可以重发，由Flink维护offset，作为状态存储

    作为sink,Flink官方的实现方法是基于两阶段提交，能保证写入的Exactly-Once，分为以下几个步骤：

  ① 开始事务（beginTransaction） 创建一个临时文件夹， 来写把数据写入到这个文件夹里面

  ② 预提交（preCommit） 将内存中缓存的数据写入文件并关闭

  ③ 正式提交（ commit） 将之前写完的临时文件放入目标目录下。 这代表着最终的数据会有一些延迟

  ④ 丢弃（abort） 丢弃临时文件

  ⑤ 若失败发生在预提交成功后， 正式提交前。 可以根据状态来提交预提交的数据， 也可删除预提交的数据。

如果下级存储不支持事务,具体实现是幂等写入，需要下级存储具有幂等性写入特性。

    比如结合HBase的rowkey的唯一性、数据的多版本，实现幂等

8 Flink分布式快照的原理?

  Flink的容错机制的核心部分是制作分布式数据流和操作算子状态的一致性快照。

  这些快照充当一致性checkpoint，系统可以在发生故障时回滚。 Flink用于制作这些快照的机制在“分布式数据流的轻量级异步快照”中进行了描述。 它受到分布式快照的标准Chandy-Lamport算法的启发，专门针对Flink的执行模型而定制。简单来说就是持续创建分布式数据流及其状态的一致快照。

  barriers在数据流源处被注入并行数据流中。快照n的barriers被插入的位置（我们称之为Sn）是快照所包含的数据在数据源中最大位置。

  例如，在Apache Kafka中，此位置将是分区中最后一条记录的偏移量。 将该位置Sn报告给checkpoint协调器（Flink的JobManager）。

  然后barriers向下游流动。当一个中间操作算子从其所有输入流中收到快照n的barriers时，它会为快照n发出barriers进入其所有输出流中。

  一旦sink操作算子（流式DAG的末端）从其所有输入流接收到barriers n，它就向checkpoint协调器确认快照n完成。

  在所有sink确认快照后，意味快照着已完成。一旦完成快照n，job将永远不再向数据源请求Sn之前的记录，因为此时这些记录（及其后续记录）将已经通过整个数据流拓扑，也即是已经被处理结束。

核心思想是在 input source 端插入 barrier， 控制 barrier 的同步来实现 snapshot 的备份和 exactly-once 语义。

9 CheckPoint

9.1 Flink 的容错机制（checkpoint）

  Checkpoint 机制是 Flink 可靠性的基石，可以保证 Flink 集群在某个算子因为某些原因(如 异常退出)出现故障时，能够将整个应用流图的状态恢复到故障之前的某一状态，保证应用流图状态的一致性。Flink 的 Checkpoint 机制原理来自“Chandy-Lamport algorithm”算法。

  每个需要 Checkpoint 的应用在启动时，Flink 的 JobManager 为其创建一个

CheckpointCoordinator(检查点协调器)，CheckpointCoordinator 全权负责本应用的快照制作。

  1. CheckpointCoordinator(检查点协调器) 周期性的向该流应用的所有source 算子发送 barrier(屏障)。

  2. 当某个 source 算子收到一个 barrier 时，便暂停数据处理过程，然后将自己的当前状态制作成快照，并保存到指定的持久化存储中，最后向CheckpointCoordinator 报告自己快照制作情况，同时向自身所有下游算子广播该 barrier，恢复数据处理

  3. 下游算子收到 barrier 之后，会暂停自己的数据处理过程，然后将自身的相关状态制作成快照，并保存到指定的持久化存储中，最后向CheckpointCoordinator 报告自身快照情况，同时向自身所有下游算子广播该 barrier，恢复数据处理。

  4. 每个算子按照步骤 3 不断制作快照并向下游广播，直到最后 barrier 传递到 sink 算子，快照制作完成。

  5. 当 CheckpointCoordinator 收到所有算子的报告之后，认为该周期的快照制作成功; 否则，如果在规定的时间内没有收到所有算子的报告，则认为本周期快照制作失败。

9.2 Flink checkpoint 与 Spark Streaming 的有什么区别或优势吗

spark streaming 的 checkpoint 仅仅是针对 driver 的故障恢复做了数据和元数据的 checkpoint。而 flink 的 checkpoint 机制 要复杂了很多，它采用的是轻量级的分布式快照，实现了每个算子的快照，及流动中的数据的快照。

9.3 Checkpoint的参数设置

  1）间隔、语义： 1min~10min，3min，语义默认精准一次

  因为一些异常原因可能导致某些barrier无法向下游传递，造成job失败，对于一些时效性要求高、精准性要求不是特别严格的指标，可以设置为至少一次。

  2）超时 ： 参考间隔， 0.5~2倍之间， 建议0.5倍

  3）最小等待间隔：上一次ck结束 到 下一次ck开始 之间的时间间隔，设置间隔的0.5倍

  4）设置保存ck：Retain

  5）失败次数：5

  6）Task重启策略（Failover）：

    固定延迟重启策略： 重试几次、每次间隔多久

    失败率重启策略： 重试次数、重试区间、重试间隔

10 CEP

  CEP全称为Complex Event Processing，复杂事件处理

  Flink CEP是在 Flink 中实现的复杂事件处理（CEP）库

  CEP 允许在无休止的事件流中检测事件模式，让我们有机会掌握数据中重要的部分

  一个或多个由简单事件构成的事件流通过一定的规则匹配，然后输出用户想得到的数据 —— 满足规则的复杂事件

11 Flink CEP 编程中当状态没有到达的时候会将数据保存在哪里？

  在流式处理中，CEP 当然是要支持 EventTime 的，那么相对应的也要支持数据的迟到现象，也就是watermark的处理逻辑。CEP对未匹配成功的事件序列的处理，和迟到数据是类似的。在 Flink CEP的处理逻辑中，状态没有满足的和迟到的数据，都会存储在一个Map数据结构中，也就是说，如果我们限定判断事件序列的时长为5分钟，那么内存中就会存储5分钟的数据，这在我看来，也是对内存的极大损伤之一。

12 Flink SQL的工作机制？

  通过Calcite对编写的 Sql 进行解析、验证、优化等操作。

  1）在Table/SQL 编写完成后，通过Calcite 中的parse、validate、rel阶段，以及Blink额外添加的convert阶段,将其先转为Operation；

  2）通过Blink Planner 的translateToRel、optimize、translateToExecNodeGraph和translateToPlan四个阶段，将Operation转换成DataStream API的 Transformation；

  3）再经过StreamJraph -> JobGraph -> ExecutionGraph等一系列流程，SQL最终被提交到集群。

13 FlinkSQL怎么对SQL语句进行优化的？

  会使用两个优化器:RBO(基于规则的优化器) 和 CBO(基于代价的优化器)

  1）RBO(基于规则的优化器)会将原有表达式裁剪掉，遍历一系列规则（Rule），只要满足条件就转换，生成最终的执行计划。一些常见的规则包括分区裁剪（Partition Prune）、列裁剪、谓词下推（Predicate Pushdown）、投影下推（Projection Pushdown）、聚合下推、limit下推、sort下推、常量折叠（Constant Folding）、子查询内联转join等。

  2）CBO(基于代价的优化器)会将原有表达式保留，基于统计信息和代价模型，尝试探索生成等价关系表达式，最终取代价最小的执行计划。CBO的实现有两种模型，Volcano模型，Cascades模型。这两种模型思想很是相似，不同点在于Cascades模型一边遍历SQL逻辑树，一边优化，从而进一步裁剪掉一些执行计划。

14 Flink提交流程、组件通讯、任务执行、内存模型

15 Flink常见的维表Join方案

  1）预加载： open（）方法，查询维表，存储下来 ==》 定时查询

   2）热存储： 存在外部系统redis、hbase等

    缓存

    异步查询： 异步io功能

   3）广播维表

   4）Temporal join：外部存储，connector创建

16 公司怎么提交的实时任务，有多少Job Manager？ 有多少TaskManager？

  1）我们使用yarn per-job模式提交任务

  2）集群默认只有一个 Job Manager。但为了防止单点故障，可以配置高可用。对于standlone模式，公司一般配置一个主 Job Manager，两个备用 Job Manager，然后结合 ZooKeeper 的使用，来达到高可用；对于yarn模式，yarn在Job Mananger故障会自动进行重启，所以只需要一个，配置的最大重启次数可以是10次。

  3）基于yarn，动态申请TaskManager的数量

17 Flink的interval join的实现原理？join不上的怎么办？

  底层调用的是keyby+connect ，处理逻辑：

  1）判断是否迟到（迟到就不处理了）

  2）每条流都存了一个Map类型的状态（key是时间戳，value是List存数据）

  3）任一条流，来了一条数据，遍历对方的map状态，能匹配上就发往join方法

  4）超过有效时间范围，会删除对应Map中的数据（不是clear，是remove）

  Interval join不会处理join不上的数据，如果需要没join上的数据，可以用 coGroup+connect算子实现，或者直接使用flinksql里的left join或right join语法。

18 Flink资源配置调优

  Flink性能调优的第一步，就是为任务分配合适的资源，在一定范围内，增加资源的分配与性能的提升是成正比的，实现了最优的资源配置后，在此基础上再考虑进行后面论述的性能调优策略。

提交方式主要是yarn-per-job，资源的分配在使用脚本提交Flink任务时进行指定。

  标准的Flink任务提交脚本（Generic CLI 模式）：

  从1.11开始，增加了通用客户端模式，参数使用-D <property=value>指定

bin/flink run \
	-t yarn-per-job \
	-d \
	-p 5 \ 指定并行度
	-Dyarn.application.queue=test \ 指定yarn队列
	-Djobmanager.memory.process.size=1024mb \ 指定JM的总进程大小
	-Dtaskmanager.memory.process.size=1024mb \ 指定每个TM的总进程大小
	-Dtaskmanager.numberOfTaskSlots=2 \ 指定每个TM的slot数
	-c com.atguigu.app.dwd.LogBaseApp /opt/module/gmall-flink/gmall-realtime-1.0-SNAPSHOT-jar-with-dependencies.jar

参数列表：https://ci.apache.org/projects/flink/flink-docs-release-1.12/deployment/config.html

18.1 内存设置

  生产资源配置：

bin/flink run \
	-t yarn-per-job \
	-d \
	-p 5 \ 指定并行度
	-Dyarn.application.queue=test \ 指定yarn队列
	-Djobmanager.memory.process.size=2048mb \ JM2~4G足够
	-Dtaskmanager.memory.process.size=6144mb \ 单个TM2~8G足够
	-Dtaskmanager.numberOfTaskSlots=2 \ 与容器核数1core：1slot或1core：2slot
	-c com.atguigu.app.dwd.LogBaseApp /opt/module/gmall-flink/gmall-realtime-1.0-SNAPSHOT-jar-with-dependencies.jar

  Flink是实时流处理，关键在于资源情况能不能抗住高峰时期每秒的数据量，通常用QPS/TPS来描述数据情况。

18.2 并行度设置

18.2.1 最优并行度计算

  开发完成后，先进行压测。任务并行度给10以下，测试单个并行度的处理上限。然后 总QPS/单并行度的处理能力 = 并行度

  不能只从QPS去得出并行度，因为有些字段少、逻辑简单的任务，单并行度一秒处理几万条数据。而有些数据字段多，处理逻辑复杂，单并行度一秒只能处理1000条数据。

最好根据高峰期的QPS压测，并行度*1.2倍，富余一些资源。

18.2.2 Source 端并行度的配置

数据源端是 Kafka，Source的并行度设置为Kafka对应Topic的分区数。

如果已经等于 Kafka 的分区数，消费速度仍跟不上数据生产速度，考虑下Kafka 要扩大分区，同时调大并行度等于分区数。

  Flink 的一个并行度可以处理一至多个分区的数据，如果并行度多于 Kafka 的分区数，那么就会造成有的并行度空闲，浪费资源。

18.2.3 Transform端并行度的配置

Keyby之前的算子

  一般不会做太重的操作，都是比如map、filter、flatmap等处理较快的算子，并行度可以和source保持一致。

Keyby之后的算子

  如果并发较大，建议设置并行度为 2 的整数次幂，例如：128、256、512；

  小并发任务的并行度不一定需要设置成 2 的整数次幂；

  大并发任务如果没有 KeyBy，并行度也无需设置为 2 的整数次幂；

18.2.4 Sink 端并行度的配置

  Sink 端是数据流向下游的地方，可以根据 Sink 端的数据量及下游的服务抗压能力进行评估。如果Sink端是Kafka，可以设为Kafka对应Topic的分区数。

  Sink 端的数据量小，比较常见的就是监控告警的场景，并行度可以设置的小一些。

  Source 端的数据量是最小的，拿到 Source 端流过来的数据后做了细粒度的拆分，数据量不断的增加，到 Sink 端的数据量就非常大。那么在 Sink 到下游的存储中间件的时候就需要提高并行度。

  另外 Sink 端要与下游的服务进行交互，并行度还得根据下游的服务抗压能力来设置，如果在 Flink Sink 这端的数据量过大的话，且 Sink 处并行度也设置的很大，但下游的服务完全撑不住这么大的并发写入，可能会造成下游服务直接被写挂，所以最终还是要在 Sink 处的并行度做一定的权衡。

18.3 RocksDB大状态调优

RocksDB 是基于 LSM Tree 实现的（类似HBase），写数据都是先缓存到内存中，所以RocksDB 的写请求效率比较高。RocksDB 使用内存结合磁盘的方式来存储数据，每次获取数据时，先从内存中 blockcache 中查找，如果内存中没有再去磁盘中查询。 优化后差不多单并行度 TPS 5000 record/s，性能瓶颈主要在于 RocksDB 对磁盘的读请求，所以当处理性能不够时，仅需要横向扩展并行度即可提高整个Job 的吞吐量。以下几个调优参数：

设置本地 RocksDB 多目录

  在flink-conf.yaml 中配置：

state.backend.rocksdb.localdir: /data1/flink/rocksdb,/data2/flink/rocksdb,/data3/flink/rocksdb

  注意：不要配置单块磁盘的多个目录，务必将目录配置到多块不同的磁盘上，让多块磁盘来分担压力。当设置多个 RocksDB 本地磁盘目录时，Flink 会随机选择要使用的目录，所以就可能存在三个并行度共用同一目录的情况。如果服务器磁盘数较多，一般不会出现该情况，但是如果任务重启后吞吐量较低，可以检查是否发生了多个并行度共用同一块磁盘的情况。

  当一个 TaskManager 包含 3 个 slot 时，那么单个服务器上的三个并行度都对磁盘造成频繁读写，从而导致三个并行度的之间相互争抢同一个磁盘 io，这样务必导致三个并行度的吞吐量都会下降。设置多目录实现多个并行度使用不同的硬盘从而减少资源竞争。

  如下所示是测试过程中磁盘的 IO 使用率，可以看出三个大状态算子的并行度分别对应了三块磁盘，这三块磁盘的 IO 平均使用率都保持在 45% 左右，IO 最高使用率几乎都是 100%，而其他磁盘的 IO 平均使用率相对低很多。由此可见使用 RocksDB 做为状态后端且有大状态的频繁读取时， 对磁盘IO性能消耗确实比较大。

  如下图所示，其中两个并行度共用了 sdb 磁盘，一个并行度使用 sdj磁盘。可以看到 sdb 磁盘的 IO 使用率已经达到了 91.6%，就会导致 sdb 磁盘对应的两个并行度吞吐量大大降低，从而使得整个 Flink 任务吞吐量降低。如果每个服务器上有一两块 SSD，强烈建议将 RocksDB 的本地磁盘目录配置到 SSD 的目录下，从 HDD 改为 SSD 对于性能的提升可能比配置 10 个优化参数更有效。

state.backend.incremental：开启增量检查点，默认false，改为true。

state.backend.rocksdb.predefined-options: SPINNING_DISK_OPTIMIZED_HIGH_MEM设置为机械硬盘+内存模式，有条件上SSD，指定为FLASH_SSD_OPTIMIZED

从Flink1.10开始，Flink默认将RocksDB的内存大小配置为每个task slot的托管内存。调试内存性能的问题主要是通过调整配置项taskmanager.memory.managed.size 或者 taskmanager.memory.managed.fraction以增加Flink的托管内存(即堆外内存)。对于更细粒度的控制，应该首先通过设置 state.backend.rocksdb.memory.managed为false，禁用自动内存管理，然后调整如下配置项：

① state.backend.rocksdb.block.cache-size: 整个 RocksDB 共享一个 block cache，读数据时内存的 cache 大小，该参数越大读数据时缓存命中率越高，默认大小为 8 MB，建议设置到 64 ~ 256 MB。

② state.backend.rocksdb.thread.num: 用于后台 flush 和合并 sst 文件的线程数，默认为 1，建议调大，机械硬盘用户可以改为 4 等更大的值。

③ state.backend.rocksdb.writebuffer.size: RocksDB 中，每个 State 使用一个 Column Family，每个 Column Family 使用独占的 write buffer，默认64MB。建议调大，调整这个参数通常要适当增加 L1 层的大小阈值max-size-level-base

④ state.backend.rocksdb.compaction.level.max-size-level-base：增大write buffer，一定需要增大level1层的阈值，该值太小会造成能存放的SST文件过少，层级变多造成查找困难，太大会造成文件过多，合并困难。建议设为 target_file_size_base（默认64MB） 的倍数，且不能太小，例如510倍，即320640MB。

⑤ state.backend.rocksdb.writebuffer.count: 每个 Column Family 对应的 writebuffer 数目，默认值是 2，对于机械磁盘来说，如果内存⾜够大，可以调大到 5 左右

⑥ state.backend.rocksdb.writebuffer.number-to-merge: 将数据从 writebuffer 中 flush 到磁盘时，需要合并的 writebuffer 数量，默认值为 1，可以调成3。

⑦ state.backend.local-recovery: 设置本地恢复，当 Flink 任务失败时，可以基于本地的状态信息进行恢复任务，可能不需要从 hdfs 拉取数据

18.4 Checkpoint设置

一般Checkpoint 时间间隔可以设置为分钟级别，例如 1 分钟、3 分钟，对于状态很大的任务每次 Checkpoint 访问 HDFS 比较耗时，可以设置为 5~10 分钟一次Checkpoint，并且调大两次 Checkpoint 之间的暂停间隔，例如设置两次Checkpoint 之间至少暂停 4或8 分钟。

  如果 Checkpoint 语义配置为 EXACTLY_ONCE，那么在 Checkpoint 过程中还会存在 barrier 对齐的过程，可以通过 Flink Web UI 的 Checkpoint 选项卡来查看 Checkpoint 过程中各阶段的耗时情况，从而确定到底是哪个阶段导致 Checkpoint 时间过长然后针对性的解决问题。

  RocksDB相关参数在1.3中已说明，可以在flink-conf.yaml指定，也可以在Job的代码中调用API单独指定，这里不再列出。

 // 使⽤ RocksDBStateBackend 做为状态后端，并开启增量 Checkpoint
 RocksDBStateBackend rocksDBStateBackend = new RocksDBStateBackend("hdfs://hadoop1:8020/flink/checkpoints", true);
 env.setStateBackend(rocksDBStateBackend);

 // 开启Checkpoint，间隔为 3 分钟
 env.enableCheckpointing(TimeUnit.MINUTES.toMillis(3));
 // 配置 Checkpoint
 CheckpointConfig checkpointConf = env.getCheckpointConfig();
 checkpointConf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
 // 最小间隔 4分钟
 checkpointConf.setMinPauseBetweenCheckpoints(TimeUnit.MINUTES.toMillis(4))
 // 超时时间 10分钟
 checkpointConf.setCheckpointTimeout(TimeUnit.MINUTES.toMillis(10));
 // 保存checkpoint
 checkpointConf.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

18.5 使用 Flink ParameterTool 读取配置

  在实际开发中，有各种环境（开发、测试、预发、生产），作业也有很多的配置：算子的并行度配置、Kafka 数据源的配置（broker 地址、topic 名、group.id）、Checkpoint 是否开启、状态后端存储路径、数据库地址、用户名和密码等各种各样的配置，可能每个环境的这些配置对应的值都是不一样的。

  如果你是直接在代码⾥⾯写死的配置，每次换个环境去运行测试作业，都要重新去修改代码中的配置，然后编译打包，提交运行，这样就要花费很多时间在这些重复的劳动力上了。在 Flink 中可以通过使用 ParameterTool 类读取配置，它可以读取环境变量、运行参数、配置文件。

ParameterTool 是可序列化的，所以你可以将它当作参数进行传递给算子的自定义函数类。

18.5.1 读取运行参数

  我们可以在Flink的提交脚本添加运行参数，格式：

  --参数名 参数值

  -参数名 参数值

在 Flink 程序中可以直接使用 ParameterTool.fromArgs(args) 获取到所有的参数，也可以通过 parameterTool.get(“username”) 方法获取某个参数对应的值。

  举例：通过运行参数指定jobname

bin/flink run \
	-t yarn-per-job \
	-d \
	-p 5 \ 指定并行度
	-Dyarn.application.queue=test \ 指定yarn队列
	-Djobmanager.memory.process.size=1024mb \ 指定JM的总进程大小
	-Dtaskmanager.memory.process.size=1024mb \ 指定每个TM的总进程大小
	-Dtaskmanager.numberOfTaskSlots=2 \ 指定每个TM的slot数
	-c com.atguigu.app.dwd.LogBaseApp /opt/module/gmall-flink/gmall-realtime-1.0-SNAPSHOT-jar-with-dependencies.jar \
	--jobname dwd-LogBaseApp  //参数名自己随便起，代码里对应上即可

  在代码里获取参数值：

ParameterTool parameterTool = ParameterTool.fromArgs(args);

String myJobname = parameterTool.get("jobname");  //参数名对应

env.execute(myJobname);

18.5.2 读取系统属性

ParameterTool 还⽀持通过 ParameterTool.fromSystemProperties() 方法读取系统属性。做个打印：

ParameterTool parameterTool = ParameterTool.fromSystemProperties();
System.out.println(parameterTool.toMap().toString());

可以得到全面的系统属性，部分结果：

18.5.3 读取配置文件

  可以使用ParameterTool.fromPropertiesFile(“/application.properties”) 读取 properties 配置文件。可以将所有要配置的地方（比如并行度和一些 Kafka、MySQL 等配置）都写成可配置的，然后其对应的 key 和 value 值都写在配置文件中，最后通过 ParameterTool 去读取配置文件获取对应的值。

18.5.4 注册全局参数

  在 ExecutionConfig 中可以将 ParameterTool 注册为全作业参数的参数，这样就可以被 JobManager 的web 端以及用户⾃定义函数中以配置值的形式访问。

StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();

env.getConfig().setGlobalJobParameters(ParameterTool.fromArgs(args));

  可以不用将ParameterTool当作参数传递给算子的自定义函数，直接在用户⾃定义的 Rich 函数中直接获取到参数值了。

env.addSource(new RichSourceFunction() { 
@Override 
public void run(SourceContext sourceContext) throws Exception {
 while (true) { 
	ParameterTool parameterTool = (ParameterTool)getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
   }
  } 

  @Override 
  public void cancel() {
  }
})

18.6 压测方式

  压测的方式很简单，先在kafka中积压数据，之后开启Flink任务，出现反压，就是处理瓶颈。相当于水库先积水，一下子泄洪。数据可以是自己造的模拟数据，也可以是生产中的部分数据。

19 反压处理

  反压（BackPressure）通常产生于这样的场景：短时间的负载高峰导致系统接收数据的速率远高于它处理数据的速率。许多日常问题都会导致反压，例如，垃圾回收停顿可能会导致流入的数据快速堆积，或遇到大促、秒杀活动导致流量陡增。反压如果不能得到正确的处理，可能会导致资源耗尽甚至系统崩溃。

反压机制是指系统能够自己检测到被阻塞的 Operator，然后自适应地降低源头或上游数据的发送速率，从而维持整个系统的稳定。Flink 任务一般运行在多个节点上，数据从上游算子发送到下游算子需要网络传输，若系统在反压时想要降低数据源头或上游算子数据的发送速率，那么肯定也需要网络传输。所以下面先来了解一下 Flink 的网络流控（Flink 对网络数据流量的控制）机制。

19.1 反压现象及定位

  Flink 的反压太过于天然了，导致无法简单地通过监控 BufferPool 的使用情况来判断反压状态。Flink 通过对运行中的任务进行采样来确定其反压，如果一个 Task 因为反压导致处理速度降低了，那么它肯定会卡在向 LocalBufferPool 申请内存块上。那么该 Task 的 stack trace 应该是这样：

java.lang.Object.wait(Native Method)
o.a.f.[...].LocalBufferPool.requestBuffer(LocalBufferPool.java:163) 
o.a.f.[...].LocalBufferPool.requestBufferBlocking(LocalBufferPool.java:133) [...]

  监控对正常的任务运行有一定影响，因此只有当 Web 页面切换到 Job 的 BackPressure 页面时，JobManager 才会对该 Job 触发反压监控。默认情况下，JobManager 会触发 100 次 stack trace 采样，每次间隔 50ms 来确定反压。Web 界面看到的比率表示在内部方法调用中有多少 stack trace 被卡在LocalBufferPool.requestBufferBlocking()，例如: 0.01 表示在 100 个采样中只有 1 个被卡在LocalBufferPool.requestBufferBlocking()。采样得到的比例与反压状态的对应关系如下：

OK: 0 <= 比例 <= 0.10

LOW: 0.10 < 比例 <= 0.5

HIGH: 0.5 < 比例 <= 1

  Task 的状态为 OK 表示没有反压，HIGH 表示这个 Task 被反压。

19.1.1 利用 Flink Web UI 定位产生反压的位置

  在 Flink Web UI 中有 BackPressure 的页面，通过该页面可以查看任务中 subtask 的反压状态，如下两图所示，分别展示了状态是 OK 和 HIGH 的场景。 排查的时候，先把operator chain禁用，方便定位。

19.1.2 利用Metrics定位反压位置

  当某个 Task 吞吐量下降时，基于 Credit 的反压机制，上游不会给该 Task 发送数据，所以该 Task 不会频繁卡在向 Buffer Pool 去申请 Buffer。反压监控实现原理就是监控 Task 是否卡在申请 buffer 这一步，所以遇到瓶颈的 Task 对应的反压⻚⾯必然会显示 OK，即表示没有受到反压。

  如果该 Task 吞吐量下降，造成该Task 上游的 Task 出现反压时，必然会存在：该 Task 对应的 InputChannel 变满，已经申请不到可用的Buffer 空间。如果该 Task 的 InputChannel 还能申请到可用 Buffer，那么上游就可以给该 Task 发送数据，上游 Task 也就不会被反压了，所以说遇到瓶颈且导致上游 Task 受到反压的 Task 对应的 InputChannel 必然是满的（这⾥不考虑⽹络遇到瓶颈的情况）。从这个思路出发，可以对该 Task 的 InputChannel 的使用情况进行监控，如果 InputChannel 使用率 100%，那么该 Task 就是我们要找的反压源。Flink 1.9 及以上版本inPoolUsage 表示 inputFloatingBuffersUsage 和inputExclusiveBuffersUsage 的总和。

  反压时，可以看到遇到瓶颈的该Task的inPoolUage为1。

19.2 反压的原因及处理

  先检查基本原因，然后再深入研究更复杂的原因，最后找出导致瓶颈的原因。下面列出从最基本到比较复杂的一些反压潜在原因。

  注意：反压可能是暂时的，可能是由于负载高峰、CheckPoint 或作业重启引起的数据积压而导致反压。如果反压是暂时的，应该忽略它。另外，请记住，断断续续的反压会影响我们分析和解决问题。

19.2.1 系统资源

  检查涉及服务器基本资源的使用情况，如CPU、网络或磁盘I/O，目前 Flink 任务使用最主要的还是内存和 CPU 资源，本地磁盘、依赖的外部存储资源以及网卡资源一般都不会是瓶颈。如果某些资源被充分利用或大量使用，可以借助分析工具，分析性能瓶颈（JVM Profiler+ FlameGraph生成火焰图）。

如何生成火焰图：http://www.54tianzhisheng.cn/2020/10/05/flink-jvm-profiler/

如何读懂火焰图：https://zhuanlan.zhihu.com/p/29952444

  ① 针对特定的资源调优Flink

  ② 通过增加并行度或增加集群中的服务器数量来横向扩展

  ③ 减少瓶颈算子上游的并行度，从而减少瓶颈算子接收的数据量（不建议，可能造成整个Job数据延迟增大）

19.2.2 垃圾收集（GC）

长时间GC暂停会导致性能问题。可以通过打印调试GC日志（通过-XX:+PrintGCDetails）或使用某些内存或 GC 分析器（GCViewer工具）来验证是否处于这种情况。

  在Flink提交脚本中,设置JVM参数，打印GC日志：

bin/flink run \
-t yarn-per-job \
-d \
-p 5 \ 指定并行度
-Dyarn.application.queue=test \ 指定yarn队列
-Djobmanager.memory.process.size=1024mb \ 指定JM的总进程大小
-Dtaskmanager.memory.process.size=1024mb \ 指定每个TM的总进程大小
-Dtaskmanager.numberOfTaskSlots=2 \ 指定每个TM的slot数
-Denv.java.opts="-XX:+PrintGCDetails -XX:+PrintGCDateStamps"
-c com.atguigu.app.dwd.LogBaseApp /opt/module/gmall-flink/gmall-realtime-1.0-SNAPSHOT-jar-with-dependencies.jar

下载GC日志的方式：

  因为是on yarn模式，运行的节点一个一个找比较麻烦。可以打开WebUI，选择JobManager或者TaskManager，点击Stdout，即可看到GC日志，点击下载按钮即可将GC日志通过HTTP的方式下载下来。

分析GC日志：

  通过 GC 日志分析出单个 Flink Taskmanager 堆总大小、年轻代、老年代分配的内存空间、Full GC 后老年代剩余大小等，相关指标定义可以去 Github 具体查看。

GCViewer地址：https://github.com/chewiebug/GCViewer

  扩展：最重要的指标是Full GC 后，老年代剩余大小这个指标，按照《Java 性能优化权威指南》这本书 Java 堆大小计算法则，设 Full GC 后老年代剩余大小空间为 M，那么堆的大小建议 3 ~ 4倍 M，新生代为 1 ~ 1.5 倍 M，老年代应为 2 ~ 3 倍 M。

19.2.3 CPU/线程瓶颈

  有时，一个或几个线程导致 CPU 瓶颈，而整个机器的CPU使用率仍然相对较低，则可能无法看到 CPU 瓶颈。例如，48核的服务器上，单个 CPU 瓶颈的线程仅占用 2％的 CPU 使用率，就算单个线程发生了 CPU 瓶颈，我们也看不出来。可以考虑使用2.2.1提到的分析工具，它们可以显示每个线程的 CPU 使用情况来识别热线程。

19.2.4 线程竞争

  与上⾯的 CPU/线程瓶颈问题类似，subtask 可能会因为共享资源上高负载线程的竞争而成为瓶颈。同样，可以考虑使用2.2.1提到的分析工具，考虑在用户代码中查找同步开销、锁竞争，尽管避免在用户代码中添加同步。

19.2.5 负载不平衡

  如果瓶颈是由数据倾斜引起的，可以尝试通过将数据分区的 key 进行加盐或通过实现本地预聚合来减轻数据倾斜的影响。

19.2.6 外部依赖

  如果发现我们的 Source 端数据读取性能比较低或者 Sink 端写入性能较差，需要检查第三方组件是否遇到瓶颈。例如，Kafka 集群是否需要扩容，Kafka 连接器是否并行度较低，HBase 的 rowkey 是否遇到热点问题。关于第三方组件的性能问题，需要结合具体的组件来分析。

20 数据倾斜

20.1 判断是否存在数据倾斜

  相同 Task 的多个 Subtask 中，个别Subtask 接收到的数据量明显大于其他 Subtask 接收到的数据量，通过 Flink Web UI 可以精确地看到每个 Subtask 处理了多少数据，即可判断出 Flink 任务是否存在数据倾斜。通常，数据倾斜也会引起反压。

20.2 数据倾斜的解决

20.2.1 keyBy 后的聚合操作存在数据倾斜

使用LocalKeyBy的思想：在 keyBy 上游算子数据发送之前，首先在上游算子的本地对数据进行聚合后再发送到下游，使下游接收到的数据量大大减少，从而使得 keyBy 之后的聚合操作不再是任务的瓶颈。 类似MapReduce 中 Combiner 的思想，但是这要求聚合操作必须是多条数据或者一批数据才能聚合，单条数据没有办法通过聚合来减少数据量。从Flink LocalKeyBy 实现原理来讲，必然会存在一个积攒批次的过程，在上游算子中必须攒够一定的数据量，对这些数据聚合后再发送到下游。

  注意：Flink是实时流处理，如果keyby之后的聚合操作存在数据倾斜，且没有开窗口的情况下，简单的使用两阶段聚合，是不能解决问题的。 因为这个时候Flink是来一条处理一条，且向下游发送一条结果，对于原来keyby的维度（第二阶段聚合）来讲，数据量并没有减少，且结果重复计算（非FlinkSQL，未使用回撤流），如下图所示：

  实现方式：以计算PV为例，keyby之前，使用flatMap实现LocalKeyby

class LocalKeyByFlatMap extends RichFlatMapFunction<String, Tuple2<String, 
 //Checkpoint 时为了保证 Exactly Once，将 buffer 中的数据保存到该 ListState 中
 private ListState<Tuple2<String, Long>> localPvStatListState;
 
 //本地 buffer，存放 local 端缓存的 app 的 pv 信息
 private HashMap<String, Long> localPvStat;
 
 //缓存的数据量大小，即：缓存多少数据再向下游发送
 private int batchSize;
 
 //计数器，获取当前批次接收的数据量
 private AtomicInteger currentSize;

 //构造器，批次大小传参
 LocalKeyByFlatMap(int batchSize){
 	this.batchSize = batchSize;
 }

 @Override
 public void flatMap(String in, Collector collector) throws Exception {
 	// 将新来的数据添加到 buffer 中
 	Long pv = localPvStat.getOrDefault(in, 0L);
 	localPvStat.put(in, pv + 1);
 	// 如果到达设定的批次，则将 buffer 中的数据发送到下游
 	if(currentSize.incrementAndGet() >= batchSize){
 		// 遍历 Buffer 中数据，发送到下游
 		for(Map.Entry<String, Long> appIdPv: localPvStat.entrySet()) {
 			collector.collect(Tuple2.of(appIdPv.getKey(), appIdPv.getValue()
 		}
 		// Buffer 清空，计数器清零
 		localPvStat.clear();
 		currentSize.set(0);
 	}
 }

 @Override
 public void snapshotState(FunctionSnapshotContext functionSnapshotConte
 	// 将 buffer 中的数据保存到状态中，来保证 Exactly Once
 	localPvStatListState.clear();
 	for(Map.Entry<String, Long> appIdPv: localPvStat.entrySet()) {
 		localPvStatListState.add(Tuple2.of(appIdPv.getKey(), appIdPv.ge
 	}
 }

 @Override
 public void initializeState(FunctionInitializationContext context) {
 	// 从状态中恢复 buffer 中的数据
 	localPvStatListState = context.getOperatorStateStore().getListState
 	new ListStateDescriptor<>("localPvStat",
 	TypeInformation.of(new TypeHint<Tuple2<String, Long>>})));
 	localPvStat = new HashMap();
 	if(context.isRestored()) {
 		// 从状态中恢复数据到 localPvStat 中
 		for(Tuple2<String, Long> appIdPv: localPvStatListState.get()){
long pv = localPvStat.getOrDefault(appIdPv.f0, 0L);
 			// 如果出现 pv != 0,说明改变了并行度，
 			// ListState 中的数据会被均匀分发到新的 subtask中
 			// 所以单个 subtask 恢复的状态中可能包含两个相同的 app 的数据
 			localPvStat.put(appIdPv.f0, pv + appIdPv.f1);
 		}
 		// 从状态恢复时，默认认为 buffer 中数据量达到了 batchSize，需要向下游发
 		currentSize = new AtomicInteger(batchSize);
 	} else {
 		currentSize = new AtomicInteger(0);
 	}
 }

}

20.2.2 keyBy 之前发生数据倾斜

  如果 keyBy 之前就存在数据倾斜，上游算子的某些实例可能处理的数据较多，某些实例可能处理的数据较少，产生该情况可能是因为数据源的数据本身就不均匀，例如由于某些原因 Kafka 的 topic 中某些 partition 的数据量较大，某些 partition 的数据量较少。对于不存在 keyBy 的 Flink 任务也会出现该情况。

  这种情况，需要让 Flink 任务强制进行shuffle。使用shuffle、rebalance 或 rescale算子即可将数据均匀分配，从而解决数据倾斜的问题。

20.2.3 keyBy 后的窗口聚合操作存在数据倾斜

  因为使用了窗口，变成了有界数据的处理，窗口默认是触发时才会输出一条结果发往下游，所以可以使用两阶段聚合的方式：

实现思路：

  第一阶段聚合：key拼接随机数前缀或后缀，进行keyby、开窗、聚合

  注意：聚合完不再是WindowedStream，要获取WindowEnd作为窗口标记作为第二阶段分组依据，避免不同窗口的结果聚合到一起

  第二阶段聚合：去掉随机数前缀或后缀，按照原来的key及windowEnd作keyby、聚合

  此外还有以下方法可以参考：

  • 在数据进入窗口前做预聚合

  • 重新设计窗口聚合的 key

21 KafkaSource调优

21.1 动态发现分区

  当 FlinkKafkaConsumer 初始化时，每个 subtask 会订阅一批 partition，但是当 Flink 任务运行过程中，如果被订阅的 topic 创建了新的 partition，FlinkKafkaConsumer 如何实现动态发现新创建的 partition 并消费呢？

在使用 FlinkKafkaConsumer 时，可以开启 partition 的动态发现。 通过 Properties指定参数开启（单位是毫秒）：

FlinkKafkaConsumerBase.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS

  该参数表示间隔多久检测一次是否有新创建的 partition。默认值是Long的最小值，表示不开启，大于0表示开启。开启时会启动一个线程根据传入的interval定期获取Kafka最新的元数据，新 partition 对应的那一个 subtask 会自动发现并从earliest 位置开始消费，新创建的 partition 对其他 subtask 并不会产生影响。

  代码如下所示：

properties.setProperty(FlinkKafkaConsumerBase.KEY_PARTITION_DISCOVERY_INTERVAL_MILLIS, 30 * 1000 + ""); 

21.2 从kafka数据源生成watermark

  Kafka单分区内有序，多分区间无序。在这种情况下，可以使用 Flink 中可识别 Kafka 分区的 watermark 生成机制。使用此特性，将在 Kafka 消费端内部针对每个 Kafka 分区生成 watermark，并且不同分区 watermark 的合并方式与在数据流 shuffle 时的合并方式相同。

在单分区内有序的情况下，使用时间戳单调递增按分区生成的 watermark 将生成完美的全局 watermark。

  可以不使用 TimestampAssigner，直接用 Kafka 记录自身的时间戳：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "hadoop1:9092,hadoop2:9092,hadoop3:9092");
properties.setProperty("group.id", "fffffffffff");


FlinkKafkaConsumer<String> kafkaSourceFunction = new FlinkKafkaConsumer<>(
                "flinktest",
                new SimpleStringSchema(),
                properties
        );

kafkaSourceFunction.assignTimestampsAndWatermarks(
	WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofMinutes(2))//从kafka数据源生成watermark
);

env.addSource(kafkaSourceFunction)

21.3 设置空闲等待

如果数据源中的某一个分区/分片在一段时间内未发送事件数据，则意味着 WatermarkGenerator 也不会获得任何新数据去生成 watermark。我们称这类数据源为空闲输入或空闲源。在这种情况下，当某些其他分区仍然发送事件数据的时候就会出现问题。 比如Kafka的Topic中，由于某些原因，造成个别Partition一直没有新的数据。由于下游算子 watermark 的计算方式是取所有不同的上游并行数据源 watermark 的最小值，则其 watermark 将不会发生变化，导致窗口、定时器等不会被触发。

  为了解决这个问题，你可以使用 WatermarkStrategy 来检测空闲输入并将其标记为空闲状态。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "hadoop1:9092,hadoop2:9092,hadoop3:9092");
properties.setProperty("group.id", "fffffffffff");


FlinkKafkaConsumer<String> kafkaSourceFunction = new FlinkKafkaConsumer<>(
                "flinktest",
                new SimpleStringSchema(),
                properties
        );

kafkaSourceFunction.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .forBoundedOutOfOrderness(Duration.ofMinutes(2))//从kafka数据源生成watermark
						.withIdleness(Duration.ofMinutes(5))//设置空闲等待
);

env.addSource(kafkaSourceFunction)

21.4 Kafka的offset消费策略

  FlinkKafkaConsumer可以调用以下API，注意与”auto.offset.reset”区分开：

  ① setStartFromGroupOffsets()：默认消费策略，默认读取上次保存的offset信息，如果是应用第一次启动，读取不到上次的offset信息，则会根据这个参数auto.offset.reset的值来进行消费数据。建议使用这个。

setStartFromEarliest()：从最早的数据开始进行消费，忽略存储的offset信息

setStartFromLatest()：从最新的数据进行消费，忽略存储的offset信息

setStartFromSpecificOffsets(Map)：从指定位置进行消费

setStartFromTimestamp(long)：从topic中指定的时间点开始消费，指定时间点之前的数据忽略

当checkpoint机制开启的时候，KafkaConsumer会定期把kafka的offset信息还有其他operator的状态信息一块保存起来。当job失败重启的时候，Flink会从最近一次的checkpoint中进行恢复数据，重新从保存的offset消费kafka中的数据（也就是说，上面几种策略，只有第一次启动的时候起作用）。

  为了能够使用支持容错的kafka Consumer，需要开启checkpoint

22 FlinkSQL调优

22.1 开启MiniBatch（提升吞吐）

  MiniBatch是微批处理，原理是缓存一定的数据后再触发处理，以减少对State的访问，从而提升吞吐并减少数据的输出量。 MiniBatch主要依靠在每个Task上注册的Timer线程来触发微批，需要消耗一定的线程调度性能。

  MiniBatch默认关闭，开启方式如下:

// 初始化table environment
TableEnvironment tEnv = ...

// 获取 tableEnv的配置对象
Configuration configuration = tEnv.getConfig().getConfiguration();

// 设置参数：
// 开启miniBatch
configuration.setString("table.exec.mini-batch.enabled", "true");
// 批量输出的间隔时间
configuration.setString("table.exec.mini-batch.allow-latency", "5 s");
// 防止OOM设置每个批次最多缓存数据的条数，可以设为2万条
configuration.setString("table.exec.mini-batch.size", "20000");

FlinkSQL官网配置参数：https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/config.html

  适用场景

  微批处理通过增加延迟换取高吞吐，如果有超低延迟的要求，不建议开启微批处理。通常对于聚合的场景，微批处理可以显著的提升系统性能，建议开启。

  注意事项：

  1）目前，key-value 配置项仅被 Blink planner 支持。

  2）1.12之前的版本有bug，开启miniBatch，不会清理过期状态，也就是说如果设置状态的TTL，无法清理过期状态。1.12版本才修复这个问题。(参考ISSUE：https://issues.apache.org/jira/browse/FLINK-17096)

22.2 开启LocalGlobal（解决常见数据热点问题）

  LocalGlobal优化将原先的Aggregate分成Local+Global两阶段聚合，即MapReduce模型中的Combine+Reduce处理模式。第一阶段在上游节点本地攒一批数据进行聚合（localAgg），并输出这次微批的增量值（Accumulator）。第二阶段再将收到的Accumulator合并（Merge），得到最终的结果（GlobalAgg）。

  LocalGlobal本质上能够靠LocalAgg的聚合筛除部分倾斜数据，从而降低GlobalAgg的热点，提升性能。结合下图理解LocalGlobal如何解决数据倾斜的问题。

  由上图可知：

  ① 未开启LocalGlobal优化，由于流中的数据倾斜，Key为红色的聚合算子实例需要处理更多的记录，这就导致了热点问题。

  ② 开启LocalGlobal优化后，先进行本地聚合，再进行全局聚合。可大大减少GlobalAgg的热点，提高性能。

LocalGlobal开启方式：

  ① LocalGlobal优化需要先开启MiniBatch，依赖于MiniBatch的参数。

  ② table.optimizer.agg-phase-strategy: 聚合策略。默认AUTO，支持参数AUTO、TWO_PHASE(使用LocalGlobal两阶段聚合)、ONE_PHASE(仅使用Global一阶段聚合)。

// 初始化table environment
TableEnvironment tEnv = ...

// 获取 tableEnv的配置对象
Configuration configuration = tEnv.getConfig().getConfiguration();

// 设置参数：
// 开启miniBatch
configuration.setString("table.exec.mini-batch.enabled", "true");
// 批量输出的间隔时间
configuration.setString("table.exec.mini-batch.allow-latency", "5 s");
// 防止OOM设置每个批次最多缓存数据的条数，可以设为2万条
configuration.setString("table.exec.mini-batch.size", "20000");
// 开启LocalGlobal
configuration.setString("table.optimizer.agg-phase-strategy", "TWO_PHASE");

判断是否生效：

  观察最终生成的拓扑图的节点名字中是否包含GlobalGroupAggregate或LocalGroupAggregate。

适用场景:

LocalGlobal适用于提升如SUM、COUNT、MAX、MIN和AVG等普通聚合的性能，以及解决这些场景下的数据热点问题。

注意事项：

  ① 需要先开启MiniBatch

  ② 开启LocalGlobal需要UDAF实现Merge方法。

22.3 开启Split Distinct（解决COUNT DISTINCT热点问题）

LocalGlobal优化针对普通聚合（例如SUM、COUNT、MAX、MIN和AVG）有较好的效果，对于COUNT DISTINCT收效不明显，因为COUNT DISTINCT在Local聚合时，对于DISTINCT KEY的去重率不高，导致在Global节点仍然存在热点。

  之前，为了解决COUNT DISTINCT的热点问题，通常需要手动改写为两层聚合（增加按Distinct Key取模的打散层）。

  从Flink1.9.0版本开始，提供了COUNT DISTINCT自动打散功能，不需要手动重写。Split Distinct和LocalGlobal的原理对比参见下图。

  举例：统计一天的UV

SELECT 
	day, 
	COUNT(DISTINCT user_id)
FROM 
	T
GROUP BY 
	day

  如果手动实现两阶段聚合：

SELECT day, SUM(cnt)
FROM (
    SELECT day, COUNT(DISTINCT user_id) as cnt
    FROM T
    GROUP BY day, MOD(HASH_CODE(user_id), 1024)
)
GROUP BY day

  第一层聚合: 将Distinct Key打散求COUNT DISTINCT。

  第二层聚合: 对打散去重后的数据进行SUM汇总。

Split Distinct开启方式：

  默认不开启，使用参数显式开启：

table.optimizer.distinct-agg.split.enabled: true //默认false

table.optimizer.distinct-agg.split.bucket-num: 1024 //Split Distinct优化在第一层聚合中，被打散的bucket数目。默认1024

// 初始化table environment
TableEnvironment tEnv = ...

// 获取 tableEnv的配置对象
Configuration configuration = tEnv.getConfig().getConfiguration();

// 设置参数：
// 开启Split Distinct
configuration.setString("table.optimizer.distinct-agg.split.enabled", "true");
// 第一层打散的bucket数目
configuration.setString("table.optimizer.distinct-agg.split.bucket-num", "1024");

判断是否生效

  观察最终生成的拓扑图的节点名中是否包含Expand节点，或者原来一层的聚合变成了两层的聚合。

适用场景

  使用COUNT DISTINCT，但无法满足聚合节点性能要求。

注意事项：

  ① 目前不能在包含UDAF的Flink SQL中使用Split Distinct优化方法。

  ② 拆分出来的两个GROUP聚合还可参与LocalGlobal优化。

  ③ 从Flink1.9.0版本开始，提供了COUNT DISTINCT自动打散功能，不需要手动重写（不用像上面的例子去手动实现）。

22.4 改写为AGG WITH FILTER语法（提升大量COUNT DISTINCT场景性能）

  在某些场景下，可能需要从不同维度来统计UV，如Android中的UV，iPhone中的UV，Web中的UV和总UV，这时，可能会使用如下CASE WHEN语法。

SELECT
 day,
 COUNT(DISTINCT user_id) AS total_uv,
 COUNT(DISTINCT CASE WHEN flag IN ('android', 'iphone') THEN user_id ELSE NULL END) AS app_uv,
 COUNT(DISTINCT CASE WHEN flag IN ('wap', 'other') THEN user_id ELSE NULL END) AS web_uv
FROM T
GROUP BY day

  在这种情况下，建议使用FILTER语法, 目前的Flink SQL优化器可以识别同一唯一键上的不同FILTER参数。如，在上面的示例中，三个COUNT DISTINCT都作用在user_id列上。此时，经过优化器识别后，Flink可以只使用一个共享状态实例，而不是三个状态实例，可减少状态的大小和对状态的访问。

  将上边的CASE WHEN替换成FILTER后，如下所示:

SELECT
 day,
 COUNT(DISTINCT user_id) AS total_uv,
 COUNT(DISTINCT user_id) FILTER (WHERE flag IN ('android', 'iphone')) AS app_uv,
 COUNT(DISTINCT user_id) FILTER (WHERE flag IN ('wap', 'other')) AS web_uv
FROM T
GROUP BY day

22.5 TopN优化

22.5.1 使用最优算法

  当TopN的输入是非更新流（例如Source），TopN只有一种算法AppendRank。当TopN的输入是更新流时（例如经过了AGG/JOIN计算），TopN有2种算法，性能从高到低分别是：UpdateFastRank 和RetractRank。算法名字会显示在拓扑图的节点名字上。

  注意：apache社区版的Flink1.12目前还没有UnaryUpdateRank，阿里云实时计算版Flink才有

UpdateFastRank ：最优算法

  需要具备2个条件：

  ① 输入流有PK（Primary Key）信息，例如ORDER BY AVG。

  ② 排序字段的更新是单调的，且单调方向与排序方向相反。例如，ORDER BY COUNT/COUNT_DISTINCT/SUM（正数）DESC。

  如果要获取到优化Plan，则您需要在使用ORDER BY SUM DESC时，添加SUM为正数的过滤条件。

  AppendFast：结果只追加，不更新

  RetractRank：普通算法，性能差

  不建议在生产环境使用该算法。请检查输入流是否存在PK信息，如果存在，则可进行UpdateFastRank优化。

22.5.2 无排名优化（解决数据膨胀问题）

  TopN语法：

SELECT *
FROM (
  SELECT *,
    ROW_NUMBER() OVER ([PARTITION BY col1[, col2..]]
    ORDER BY col1 [asc|desc][, col2 [asc|desc]...]) AS rownum
  FROM table_name)
WHERE rownum <= N [AND conditions]

数据膨胀问题：

  根据TopN的语法，rownum字段会作为结果表的主键字段之一写入结果表。但是这可能导致数据膨胀的问题。例如，收到一条原排名9的更新数据，更新后排名上升到1，则从1到9的数据排名都发生变化了，需要将这些数据作为更新都写入结果表。这样就产生了数据膨胀，导致结果表因为收到了太多的数据而降低更新速度。

  使用方式

TopN的输出结果无需要显示rownum值，仅需在最终前端显式时进行1次排序，极大地减少输入结果表的数据量。只需要在外层查询中将rownum字段裁剪掉即可

// 最外层的字段，不写 rownum
SELECT col1, col2, col3
FROM (
 SELECT col1, col2, col3
   ROW_NUMBER() OVER ([PARTITION BY col1[, col2..]]
   ORDER BY col1 [asc|desc][, col2 [asc|desc]...]) AS rownum
 FROM table_name)
WHERE rownum <= N [AND conditions]

  在无rownum的场景中，对于结果表主键的定义需要特别小心。如果定义有误，会直接导致TopN结果的不正确。 无rownum场景中，主键应为TopN上游GROUP BY节点的KEY列表。

22.5.3 增加TopN的Cache大小

  TopN为了提升性能有一个State Cache层，Cache层能提升对State的访问效率。TopN的Cache命中率的计算公式为：

cache_hit = cache_size*parallelism/top_n/partition_key_num

  例如，Top100配置缓存10000条，并发50，当PatitionBy的key维度较大时，例如10万级别时，Cache命中率只有10000*50/100/100000=5%，命中率会很低，导致大量的请求都会击中State（磁盘），性能会大幅下降。因此当PartitionKey维度特别大时，可以适当加大TopN的CacheS ize，相对应的也建议适当加大TopN节点的Heap Memory。

使用方式

// 初始化table environment
TableEnvironment tEnv = ...

// 获取 tableEnv的配置对象
Configuration configuration = tEnv.getConfig().getConfiguration();

// 设置参数：
// 默认10000条，调整TopN cahce到20万，那么理论命中率能达200000*50/100/100000 = 100%
configuration.setString("table.exec.topn.cache-size", "200000");

  注意：目前源码中标记为实验项，官网中未列出该参数

22.5.4 PartitionBy的字段中要有时间类字段

  例如每天的排名，要带上Day字段。否则TopN的结果到最后会由于State ttl有错乱。

22.5.5 优化后的SQL示例

insert
  into print_test
SELECT
  cate_id,
  seller_id,
  stat_date,
  pay_ord_amt  --不输出rownum字段，能减小结果表的输出量（无排名优化）
FROM (
    SELECT
      *,
      ROW_NUMBER () OVER (
        PARTITION BY cate_id,
        stat_date  --注意要有时间字段，否则state过期会导致数据错乱（分区字段优化）
        ORDER
          BY pay_ord_amt DESC  --根据上游sum结果排序。排序字段的更新是单调的，且单调方向与排序方向相反（走最优算法）
      ) as rownum  
    FROM (
        SELECT
          cate_id,
          seller_id,
          stat_date,
          --重点。声明Sum的参数都是正数，所以Sum的结果是单调递增的，因此TopN能使用优化算法，只获取前100个数据（走最优算法）
          sum (total_fee) filter (
            where
              total_fee >= 0
          ) as pay_ord_amt
        FROM
          random_test
        WHERE
          total_fee >= 0
        GROUP
          BY cate_name,
          seller_id,
          stat_date
      ) a
    WHERE
      rownum <= 100
  );

22.6 高效去重方案

  由于SQL上没有直接支持去重的语法，还要灵活的保留第一条或保留最后一条。因此我们使用了SQL的ROW_NUMBER OVER WINDOW功能来实现去重语法。去重本质上是一种特殊的TopN。

22.6.1 保留首行的去重策略（Deduplicate Keep FirstRow）

  保留KEY下第一条出现的数据，之后出现该KEY下的数据会被丢弃掉。因为STATE中只存储了KEY数据，所以性能较优，示例如下：

SELECT *
FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b ORDER BY proctime) as rowNum
  FROM T
)
WHERE rowNum = 1

  以上示例是将T表按照b字段进行去重，并按照系统时间保留第一条数据。Proctime在这里是源表T中的一个具有Processing Time属性的字段。如果按照系统时间去重，也可以将Proctime字段简化PROCTIME()函数调用，可以省略Proctime字段的声明。

22.6.2 保留末行的去重策略（Deduplicate Keep LastRow）

  保留KEY下最后一条出现的数据。保留末行的去重策略性能略优于LAST_VALUE函数，示例如下：

SELECT *
FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY b, d ORDER BY rowtime DESC) as rowNum
  FROM T
)
WHERE rowNum = 1

  以上示例是将T表按照b和d字段进行去重，并按照业务时间保留最后一条数据。Rowtime在这里是源表T中的一个具有Event Time属性的字段。

22.7 高效的内置函数

22.7.1 使用内置函数替换自定义函数

  Flink的内置函数在持续的优化当中，请尽量使用内部函数替换自定义函数。使用内置函数好处：

  ① 优化数据序列化和反序列化的耗时。

  ② 新增直接对字节单位进行操作的功能。

支持的系统内置函数：https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/functions/systemFunctions.html

22.7.2 LIKE操作注意事项

  ① 如果需要进行StartWith操作，使用LIKE ‘xxx%’。

  ② 如果需要进行EndWith操作，使用LIKE ‘%xxx’。

  ③ 如果需要进行Contains操作，使用LIKE ‘%xxx%’。

  ④ 如果需要进行Equals操作，使用LIKE ‘xxx’，等价于str = ‘xxx’。

  ⑤ 如果需要匹配 _ 字符，请注意要完成转义LIKE ‘%seller/id%’ ESCAPE ‘/’。_在SQL中属于单字符通配符，能匹配任何字符。如果声明为 LIKE ‘%seller_id%’，则不单会匹配seller_id还会匹配seller#id、sellerxid或seller1id 等，导致结果错误。

22.7.3 慎用正则函数（REGEXP）

  正则表达式是非常耗时的操作，对比加减乘除通常有百倍的性能开销，而且正则表达式在某些极端情况下可能会进入无限循环，导致作业阻塞。建议使用LIKE。正则函数包括：

  ① REGEXP

  ② REGEXP_EXTRACT

  ③ REGEXP_REPLACE

22.8 指定时区

  本地时区定义了当前会话时区id。当本地时区的时间戳进行转换时使用。在内部，带有本地时区的时间戳总是以UTC时区表示。但是，当转换为不包含时区的数据类型时(例如TIMESTAMP, TIME或简单的STRING)，会话时区在转换期间被使用。为了避免时区错乱的问题，可以参数指定时区。

// 初始化table environment
TableEnvironment tEnv = ...

// 获取 tableEnv的配置对象
Configuration configuration = tEnv.getConfig().getConfiguration();

// 设置参数：
// 指定时区
configuration.setString("table.local-time-zone", "Asia/Shanghai");

22.9 设置参数总结

  总结以上的调优参数，代码如下：

// 初始化table environment
TableEnvironment tEnv = ...

// 获取 tableEnv的配置对象
Configuration configuration = tEnv.getConfig().getConfiguration();

// 设置参数：
// 开启miniBatch
configuration.setString("table.exec.mini-batch.enabled", "true");
// 批量输出的间隔时间
configuration.setString("table.exec.mini-batch.allow-latency", "5 s");
// 防止OOM设置每个批次最多缓存数据的条数，可以设为2万条
configuration.setString("table.exec.mini-batch.size", "20000");
// 开启LocalGlobal
configuration.setString("table.optimizer.agg-phase-strategy", "TWO_PHASE");
// 开启Split Distinct
configuration.setString("table.optimizer.distinct-agg.split.enabled", "true");
// 第一层打散的bucket数目
configuration.setString("table.optimizer.distinct-agg.split.bucket-num", "1024");
// TopN 的缓存条数
configuration.setString("table.exec.topn.cache-size", "200000");
// 指定时区
configuration.setString("table.local-time-zone", "Asia/Shanghai");

23 Flink如何保证 Exactly-once 语义

  Flink 通过实现两阶段提交和状态保存来实现端到端的一致性语义。分为以下几个步骤：

  ① 开始事务（beginTransaction）创建一个临时文件夹，来写把数据写入到这个文件夹里面

  ② 预提交（preCommit）将内存中缓存的数据写入文件并关闭

  ③ 正式提交（commit）将之前写完的临时文件放入目标目录下。这代表着最终的数据会有一些延迟

  ④ 丢弃（abort）丢弃临时文件

  ⑤ 若失败发生在预提交成功后，正式提交前。可以根据状态来提交预提交的数据，也可删除预提交的数据。

24 如果下级存储不支持事务，Flink 怎么保证exactly-once

  端到端的 exactly-once 对 sink 要求比较高，具体实现主要有幂等写入和事务性写入两种方式。

幂等写入的场景依赖于业务逻辑，更常见的是用事务性写入。而事务性写入又有预写日志（WAL）和两阶段提交（2PC）两种方式。

如果外部系统不支持事务，那么可以用预写日志的方式，把结果数据先当成状态保存，然后在收到 checkpoint 完成的通知时，一次性写入 sink 系统。

25 Operator Chains（算子链）这个概念你了解吗

  为了更高效地分布式执行，Flink 会尽可能地将 operator 的 subtask 链接（chain）在一起形成 task。每个 task 在一个线程中执行。将 operators 链接成 task 是非常有效的优化：它能减少线程之间的切换，减少消息的序列化/反序列化，减少数据在缓冲区的交换，减少了延迟的同时提高整体的吞吐量。 这就是我们所说的算子链。

两个 operator chain 在一起的的条件：

  上下游的并行度一致

  下游节点的入度为 1 （也就是说下游节点没有来自其他节点的输入）

  上下游节点都在同一个 slot group 中（下面会解释 slot group）

  下游节点的 chain 策略为 ALWAYS（可以与上下游链接， map、 flatmap、 filter 等默认是 ALWAYS）

  上游节点的 chain 策略为 ALWAYS 或 HEAD（只能与下游链接， 不能与上游链接，Source 默认是 HEAD）

  两个节点间数据分区方式是 forward（参考理解数据流的分区）

  用户没有禁用 chain

26 Flink 的内存管理是如何做的

Flink 并不是将大量对象存在堆上，而是将对象都序列化到一个预分配的内存块上。 此外，Flink 大量的使用了堆外内存。如果需要处理的数据超出了内存限制，则会将部分数据存储到硬盘上。 Flink 为了直接操作二进制数据实现了自己的序列化框架。理论上 Flink 的内存管理分为三部分：

  Network Buffers： 这个是在 TaskManager 启动的时候分配的， 这是一组用于缓存网络数据的内存， 每个块是32K， 默认分配 2048个， 可以通过“taskmanager.network.numberOfBuffers”修改

  Memory Manage pool： 大量的 Memory Segment 块， 用于运行时的算法（ Sort/Join/Shuffle等） ， 这部分启动的时候就会分配 ， 内存的分配支持预分配和 lazy load，默认懒加载的方式。

  User Code， 这部分是除了 Memory Manager 之外的内存用于 User code 和 TaskManager

本身的数据结构。

27 Flink 对于迟到数据是怎么处理的

  Flink 中 WaterMark 和 Window 机制解决了流式数据的乱序问题，对于因为延迟而顺序有误的数据，可以根据 EventTime 进行业务处理，对于延迟的数据 Flink也有自己的解决办法，主要的办法是给定一个允许延迟的时间，在该时间范围内仍可以接受处理延迟数据

  设置允许延迟的时间是通过 allowedLateness(lateness: Time)设置；

  保存延迟数据则是通过 sideOutputLateData(outputTag: OutputTagT)；

  获取延迟数据是通过 DataStream.getSideOutput(tag: OutputTagX)获取。

28 Flink 中 Task 如何做到数据交换

  在一个 Flink Job 中，数据需要在不同的 task 中进行交换，整个数据交换是有 TaskManager 负责的，TaskManager 的网络组件首先从缓冲 buffer 中收集records，然后再发送。Records 并不是一个一个被发送的，是积累一个批次再发送，batch 技术可以更加高效的利用网络资源。

  Flink 为了避免 JVM 的固有缺陷例如 java 对象存储密度低， FGC 影响吞吐和响应等，实现了自主管理内存。 MemorySegment 就是 Flink 的内存抽象。 默认情况下， 一个MemorySegment 可以被看做是一个 32kb 大的内存块的抽象。 这块内存既可以是 JVM 里的一个 byte[]， 也可以是堆外内存（DirectByteBuffer） 。 在 MemorySegment 这个抽象之上，Flink 在数据从 operator 内的数据对象在向 TaskManager 上转移， 预备被发给下个节点的过程中， 使用的抽象或者说内存对象是 Buffer。 对接从 Java 对象转为 Buffer 的中间对象是另一个抽象 StreamRecord。

29 介绍下 Flink 的序列化

  Flink 摒弃了 Java 原生的序列化方法，以独特的方式处理数据类型和序列化，包含自己的类型描述符，泛型类型提取和类型序列化框架。

  TypeInformation 是所有类型描述符的基类。它揭示了该类型的一些基本属性，并且可以生成序列化器。

  TypeInformation 支持以下几种类型：

  • BasicTypeInfo: 任意 Java 基本类型或 String 类型

  • BasicArrayTypeInfo: 任意 Java 基本类型数组或 String 数组

  • WritableTypeInfo: 任意 Hadoop Writable 接口的实现类

  • TupleTypeInfo: 任意的 Flink Tuple 类型(支持 Tuple1 to Tuple25)。 Flink tuples 是固定长度固定类型的 Java Tuple 实现

  • CaseClassTypeInfo: 任意的 Scala CaseClass(包括 Scala tuples)

  • PojoTypeInfo: 任意的 POJO (Java or Scala)，例如，Java 对象的所有成员变量，要么是 public 修饰符定义，要么有 getter/setter 方法

  • GenericTypeInfo: 任意无法匹配之前几种类型的类

30 Flink 海量数据高效去重

  1. 基于状态后端。

  2. 基于 HyperLogLog：不是精准的去重。

  3. 基于布隆过滤器（BloomFilter）；快速判断一个 key 是否存在于某容器，不存在就直接返回。

  4. 基于 BitMap；用一个 bit 位来标记某个元素对应的 Value，而 Key 即是该元素。由于采用了 Bit 为单位来存储数据，因此可以大大节省存储空间。

  5. 基于外部数据库；选择使用 Redis 或者 HBase 存储数据，我们只需要设计好存储的 Key 即可，不需要关心 Flink 任务重启造成的状态丢失问题。

31 Flink 常用的算子有哪些

  1. 数据读取， 这是 Flink 流计算应用的起点， 常用算子有：

    从内存读： fromElements

    从文件读： readTextFile

    Socket 接入 ： socketTextStream

    自定义读取： createInput

  2. 处理数据的算子， 常用的算子包括： Map（单输入单输出） 、 FlatMap（单 输入、 多输出） 、 Filter（过滤） 、 KeyBy（分组） 、 Reduce（聚合） 、 Window（窗口） 、 Connect（连接） 、 Split（分割） 等。

32 Flink 的 kafka 连接器有什么特别的地方？

  Flink 源码中有一个独立的 connector 模块， 所有的其他 connector 都依赖于此模块， Flink在 1.9 版本发布的全新 kafka 连接器， 摒弃了之前连接不同版本的 kafka 集群需要依赖不同版本的 connector 这种做法， 只需要依赖一个 connector 即可。

33 Flink Job 的提交流程

  用户提交的 Flink Job 会被转化成一个 DAG 任务运行， 分别是： StreamGraph、JobGraph、ExecutionGraph， Flink 中 JobManager 与 TaskManager， JobManager 与 Client 的交互是基于Akka 工具包的， 是通过消息驱动。 整个 Flink Job 的提交还包含着 ActorSystem 的创建，JobManager 的启动， TaskManager 的启动和注册。

34 Flink 所谓"三层图"结构是哪几个"图"？

  一个 Flink 任务的 DAG 生成计算图大致经历以下三个过程：

  StreamGraph 最接近代码所表达的逻辑层面的计算拓扑结构， 按照用户代码的执行顺序向StreamExecutionEnvironment 添加 StreamTransformation 构成流式图。

  JobGraph 从 StreamGraph 生成， 将可以串联合并的节点进行合并， 设置节点之间的边，安排资源共享 slot 槽位和放置相关联的节点， 上传任务所需的文件， 设置检查点配置等。 相当于经过部分初始化和优化处理的任务图。

  ExecutionGraph 由 JobGraph 转换而来， 包含了任务具体执行所需的内容， 是最贴近底层实现的执行图。

35 Map和FlatMap有什么区别？

  在Flink中，map()和flatMap()都是用来对数据集中的元素进行转换的操作，它们和Java中的map()和flatMap()的含义和用法基本一致。不同的是，在Flink中，map()和flatMap()都是通过实现MapFunction和FlatMapFunction接口来完成的。

map()函数是将输入元素转换为一个输出元素的函数，即每个输入元素只能映射为一个输出元素。因此，map()适用于将一个数据集中的元素逐一转换为另一个数据集的元素的场景。 例如，我们可以使用MapFunction将一个字符串转换为大写形式：

public static final class UpperCaseMapFunction implements MapFunction<String, String> {
    @Override
    public String map(String value) {
        return value.toUpperCase();
    }
}

DataStream<String> input = ...;
DataStream<String> output = input.map(new UpperCaseMapFunction());

flatMap()函数是将输入元素转换为零个或多个输出元素的函数，即每个输入元素可以映射为零个或多个输出元素。因此，flatMap()适用于将一个数据集中的元素拆分为多个元素的场景。 例如，我们可以使用FlatMapFunction将一个字符串按照空格进行分割，并输出每个单词：

DataStream<String> input = ...;
DataStream<String> output = input.flatMap(new FlatMapFunction<String, String>() {
    @Override
    public void flatMap(String value, Collector<String> out) {
        for (String word : value.split(" ")) {
            out.collect(word);
        }
    }
});

在flatMap()中，我们需要将每个单词使用Collector输出，因为flatMap()可以生成多个输出元素。而Map中则使用return 返回处理后字符，因为只生成一个元素。

  综上所述，map()和flatMap()的区别在于它们的输出类型不同，因此适用于不同的场景。map()适用于将一个数据集中的元素逐一转换为另一个数据集的元素的场景，flatMap()适用于将一个数据集中的元素拆分为多个元素的场景。

36 Flink重启策略

37 Flink侧输出流

38 自定义Function

39 Flink 的JOIN操作