Flink

有界数据

• 离线的计算

  • 效率：硬件相同的情况下：时间

无界数据

• 实时的计算

  • 效率：硬件环境相同的情况下，吞吐量：数据处理的数量、延迟：结果输出的时间—数据接收的时间

滚动窗口

• 数量–时间
• 不重复

滑动窗口

• 可以重复
• 滑动的间隔–窗口的宽度

会话窗口

• 不会重复
• 基于某一个用户的未 *** 作时间

第一代: MapReducer

• 批处理：Mapper, Reducer
• Hadoop的MapReducer将计算分为两个阶段, 分别为Map和Reducer. 对于上层应用来说, 就不得不想方设法去拆分算法, 甚至于不得不在上层应用实现多个Job的串联, 以完成一个完整的算法, 例如迭代计算.

第二代: DAG框架 (Tez) + MapReducer

• 批处理 1个Tez = MR (1) + MR (2) + … + MR (n) 相比MR效率有所提升

第三代: Spark

• 批处理, 流处理, SQL高层API支持 自带DAG 内存迭代计算, 性能较之前大幅提升

第四代: Flink

• 批处理, 流处理, SQL高层API支持 自带DAG 流式计算性能更高, 可靠性更高

分布式、高性能、随时可用以及准确的流处理应用程序打造的开源流处理框架

框架和分布式处理引擎，用于对无界和有界数据流进行有状态计算。Flink 被设计在所有常见的集群环境中运行，以内存执行速度和任意规模来执行计算。

Flink流处理的特点

• 同时支持高吞吐，低延迟，高性能

• 支持事件时间（Event Time） 概念

  • Process TIme,大多数框架窗口计算采用的都是系统时间（process time），也是事件传
    输到计算框架处理时，系统主机的当前时间。
  • Event Time，Flink支持基于事件时间（Event Time）语义进行窗口计算，也就是使用事
    件产生的时间，这种基于事件驱动的机制使得事件即使乱序到达，流系统也能够计算出
    精确的结果，保持了事件原本产生时的时序性，尽可能避免网络传输或硬件系统的影
    响。

• 支持有状态计算

• 支持高度灵活的窗口（Window） *** 作。

• 基于轻量级的分布式快照（CheckPoint）来实现容错

• 基于JVM实现独立的内存管理

• Save Point保存点

  • CheckPoint是自动的，做容错。Save Point是手动的，处理程序升级。

有界数据

• 具有起始时间和截止时间。它可以在执行任何的计算之前，先通过摄取所有数据后再来处理有
  界流。处理有界流不需要有序摄取，因为可以对有界数据集进行排序。有界流的处理也称为批
  处理。

• 特点

  • 有定义流的开始，也有定义流的结束。
  • 有界流可以在摄取所有数据后再进行计算。
  • 有界流所有数据可以被排序，所以并不需要有序摄取。
  • 有界流处理通常被称为批处理

无界数据

• 它有开始时间但没有截止时间，它们在生成时提供数据，但不会被终止。无界流必须连续处理
  数据，即必须在摄取事件后立即处理事件。它无法等待所有输入数据到达，因为输入是无界
  的，如果是这样，在任何时间点都不会完成。处理无界数据通常要求以特定顺序摄取事件，例
  如事件发生的顺序，以便能够推断结果完整性。

• 特点

  • 有定义流的开始，但没有定义流的结束。它们会无休止地产生数据。
  • 无界流的数据必须持续处理，即数据被摄取后需要立刻处理。我们不能等到所有数据都到达再处理，因为输入是无限的，在任何时候输入都不会完成。
  • 处理无界数据通常要求以特定顺序摄取事件，例如事件发生的顺序，以便能够推断结果的完整性。

• Flink 擅长精确的时间控制和状态化，使得 Flink 的运行时(runtime)能够运行任何处理无界流的应用。

SQL&Table API

• SQL 构建在Table 之上，都需要构建Table 环境。
• 不同的类型的Table 构建不同的Table 环境中。
• Table 可以与DataStream或者DataSet进行相互转换。
• Streaming SQL不同于存储的SQL，最终会转化为流式执行计划

DataStream/DataSet API

• ProcessFunction多了一些算子。DataStream API为许多通用的流处理 *** 作提供了原语。
• DataSet API 是批处理API，处理有限的数据集。
• DataStream API是流处理API，处理无限的数据集。

Stateful Stream processing

• ProcessFunction是Flink最底层的接口。
• ProcessFunction可以处理一或者两条输入数据流中的单个事件或者归入一个特定窗口内的多个事件。
• 它提供了对时间和状态的细粒度控制。
• 虽然灵活性高，但开发比较复杂，需要具备一定的编码能力。

扩展库

• 复杂事件处理CEP，Gelly做图计算的，是一个可扩展的图形处理和分析库。

Flink运行时架构主要包括四个不同的组件，它们会在运行流处理应用程序时协同工作：

• 作业管理器（JobManager）
• 资源管理器（ResourceManager）
• 任务管理器（TaskManager）
• 分发器（Dispatcher）

JobManager

• 管理任务，申请资源（slot插槽）

ResourceManager

• 管理资源，掌握各个节点当前主机的情况

TaskManager

• 管理任务，申请资源

Dispatcher

• 提供Flink外部访问接口，用于接受客户端的请求

Yarn-Session模式

• 所有的Flink任务共用一个资源，当开启一个新的Job的时候，开始分配Container

Single-job模式

• 每次Job都会形成一个新的yarn任务，任务和任务之间相互独立

查看执行图

图形解释

• Flink 中的执行图可以分成四层：StreamGraph -> JobGraph -> ExecutionGraph -> 物理执行图。

  • StreamGraph：

    • 根据用户通过 Stream API 编写的代码生成的最初的图。用来表示程序的拓扑结构。

  • JobGraph：

    • StreamGraph经过优化后生成了 JobGraph，提交给 JobManager 的数据结构。主要的优化为，将多个符合条件的节点 chain 在一起作为一个节点，这样可以减少数据在节点之间流动所需要的序列化/反序列化/传输消耗。

  • ExecutionGraph：

    • JobManager 根据 JobGraph 生成ExecutionGraph。
    • ExecutionGraph是JobGraph的并行化版本，是调度层最核心的数据结构。

  • 物理执行图：

    • JobManager 根据 ExecutionGraph 对 Job 进行调度后，在各个TaskManager 上部署Task 后形成的“图”
    • 并不是一个具体的数据结构。

一个TaskManager可以同时执行多个任务（tasks）。

• 这些任务可以是同一个算子的子任务（数据并行）
• 这些任务可以是来自不同算子（任务并行）
• 这些任务可以是另一个不同应用程序（作业并行）。

TaskManager提供了一定数量的处理插槽（processing slots），用于控制可以并行执行的任务数。

一个slot可以执行应用的一个分片，也就是应用中每一个算子的一个并行任务。

Slot对资源的隔离仅仅是对内存进行隔离，策略是均分

• 比如taskmanager的管理内存是3GB，假如有两个slot，那么每个slot就仅仅有1.5GB内存可用。

插槽的数量通常与每个TaskManager的可用CPU内核数成比例。一般情况下你的slot数是你每个TM的cpu的核数

算子级别

• setParallelism()

执行环境级别

客户端级别

系统级别

Flink出于分布式执行的目的，将operator的subtask链接在一起形成task。

每个task在一个线程中执行。

将operators链接成task是非常有效的优化：

• 它可以减少线程与线程间的切换和数据缓冲的开销，并在降低延迟的同时提高整体吞吐量。

链接的行为可以在编程API中进行指定开启 *** 作链（默认） 和 禁用 *** 作链的

Processing time（处理时间）

• 处理时间是指当前机器处理该条事件的时间。
• 它是当数据流入到具体某个算子时候相应的系统时间。他提供了最小的延时和最佳的性能。但是在分布式和异步环境中，处理时间不能提供确定性，因为它对事件到达系统的速度和数据流在系统的各个operator之间处理的速度很敏感。

Event Time(事件时间)

• 事件时间是每个事件在其生产设备上发生的时间。此时间通常在进入Flink之前嵌入到记录中，并且可以从每个记录中提取该事件时间戳。事件时间对于乱序、延时、或者数据重放等情况，都能给出正确的结间水位线（wartermark），这是指示事件时间进度的机制。水位线机制在后面部分中会描述。
• 果。事件时间依赖于事件本身，而跟物理时钟没有关系。基于事件时间的程序必须指定如何生成事件时事件时间处理通常存在一定的延时，因此需要为延时和无序的事件等待一段时间。因此，使用事件时间编程通常需要与处理时间相结合。

Ingestion time(摄入时间)

• 摄入时间是数据进入Flink框架的时间，是在Source Operator中设置的。ProcessingTime相比可以提供更可预测的结果，因为摄入时间的时间戳比较稳定(在源处只记录一次)，同一数据在流经不同窗口 *** 作时将使用相同的时间戳，而对于ProcessingTime同一数据在流经不同窗口算子会有不同的处理时间戳。

设置时间特性

• streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

其中一种方式为在数据源完成的，即利用SourceFunction在应用读入数据流的时候分配时间戳与水位线。

另一种方式是通过实现接口的自定义函数，该方式又包括两种实现方式：

• 一种为周期性生成水位线，即实现AssignerWithPeriodicWatermarks接口，周期性的生成水位线
• 另一种为定点生成水位线，即实AssignerWithPunctuatedWatermarks接口。每条数据都会产生水位线

现实中很难生成一个完美的水位线，水位线就是在延迟与准确性之前做的一种权衡。那么，如果生成的水位线过于紧迫，即水位线可能会大于后来数据的时间戳，这就意味着数据有延迟，关于延迟数据的处理，Flink提供了一些机制，具体如下：

• 直接将迟到的数据丢弃
• 将迟到的数据输出到单独的数据流中，即使用sideOutputLateData(new OutputTag<>()）实现侧输出
• 根据迟到的事件更新并发出结果

基本转换算子

• map
• flatmap
• filter

键控流转换算子

• keyby

  • 滚动聚合

    • sum()：在输入流上对指定的字段做滚动相加 *** 作。
    • min()：在输入流上对指定的字段求最小值。
    • max()：在输入流上对指定的字段求最大值。
    • minBy()：在输入流上针对指定字段求最小值，并返回包含当前观察到的最小值的事件。
    • maxBy()：在输入流上针对指定字段求最大值，并返回包含当前观察到的最大值的事件。

  • Reduce

    • 滚动聚合的泛化实现

多流转换算子

• union
• connect

分布式转换算子

• shuffle

  • 随机

• rebalance

  • 负载均衡

• rescale

  • 将数据发送到接下来的并行运行的任务中的一部分任务中

• broadcast()

  • 将输入流的所有数据复制并发送到下游算子的所有并行任务中去。

• global()

  • 将所有的输入流数据都发送到下游算子的第一个并行任务中去。这个 *** 作需要很谨慎，因为将所有数据发送到同一个task，将会对应用程序造成很大的压力。

Primitives（原始数据类型）

Java和Scala的Tuples（元组）

Scala的样例类

POJO类型

• 公有类
• 无参数的公有构造器
• 所有的字段都是公有的，可以通过getters和setters访问。
• 所有字段的数据类型都必须是Flink支持的数据类型。
• Types.POJO(Person.class);

一些特殊的类型

DataStream API提供的所有转换 *** 作函数，都拥有它们的“富”版本，并且我们在使用常规函数或者匿名函数的地方来使用富函数。

• RichMapFunction
• RichFlatMapFunction
• RichFilterFunction

使用富函数时可以实现额外的方法：

• open()方法是rich function的初始化方法

  • 当一个算子例如map或者filter被调用之前open()会被调用。
  • open()函数通常用来做一些只需要做一次即可的初始化工作。

• close()方法是生命周期中的最后一个调用的方法

  • 通常用来做一些清理工作。

• getRuntimeContext()方法提供了函数的RuntimeContext的一些信息

  • 例如函数执行的并行度，当前子任务的索引，当前子任务的名字。同时还它还包含了访问分区状态的方法

算子的类型

• operator State

  • ListCheckPoint
  • snapshotState–拍摄快照
  • restartState–恢复快照拍摄的数据

• Keyed State

  • 实现Rich Function

算例的管理

• managed state

  • flink管理

• raw state

  • 自己管理

保存算子的当前状态，当集群出现故障的时候可以进行恢复，默认情况下需要手动开启，flink可以恢复检查点的数据，数据源也可以重复被消费，在流式计算中，算子的状态非常宝贵，因为有的状态很难进行重现，分布式快照：集群中的算子可以并行的去拍摄快照，不需要等待一起去拍

• 栅栏barriers

单个模式

• 一个模式可以是一个单例或者循环模式。

  • 单例模式只接受一个事件
  • 循环模式可以接受多个事件。

• 在FlinkCEP中，你可以通过这些方法指定循环模式

  • pattern.oneOrMore() ，指定期望一个给定事件出现一次或者多次的模式
  • pattern.times(#ofTimes) ，指定期望一个给定事件出现特定次数的模式
  • pattern.times(#fromTimes, #toTimes) ，指定期望一个给定事件出现次数在一个最小值和最大值中间的模式
  • pattern.greedy() 方法让循环模式变成贪心的

• 条件

  • 对每个模式你可以指定一个条件来决定一个进来的事件是否被接受进入这个模式
  • 指定判断事件属性的条件可以通过 pattern.where() 、 pattern.or() 或者pattern.until() 方法
  • IterativeCondition 或者 SimpleCondition 。

组合模式

• 连续策略

  • 1. 严格连续: 期望所有匹配的事件严格的一个接一个出现，中间没有任何不匹配的事件。
  • 2. 松散连续: 忽略匹配的事件之间的不匹配的事件。
  • 3. 不确定的松散连续: 更进一步的松散连续，允许忽略掉一些匹配事件的附加匹配。

• 指定模式之间的连续策略的方法

  • 1. next() ，指定严格连续，
  • 2. followedBy() ，指定松散连续，
  • 3. followedByAny() ，指定不确定的松散连续。
  • 4. notNext() ，如果不想后面直接连着一个特定事件
  • 5. notFollowedBy() ，如果不想一个特定事件发生在两个事件之间的任何地方。
  • 6. pattern.within() 方法指定一个模式应该在10秒内发生

循环模式

• oneOrMore() 和 times() ，默认是松散连续。如果想使用严格连续，你需要使用 consecutive()
  方法明确指定， 如果想使用不确定松散连续，你可以使用 allowCombinations() 方法。

匹配后跳过策略

• AfterMatchSkipStrategy

  • NO_SKIP

    • 所有成功的匹配结果都会被保留

  • SKIP_TO_NEXT

    • 丢弃所有的子集

  • SKIP_PAST_LAST_EVENT

    • 每种开头相同的保留一个

  • SKIP_TO_FIRST

    • 只保留第一个开头的所有组合

  • SKIP_TO_LAST

    • 用上个模式的结尾作为本次匹配的开头

检测模式

• 在指定了要寻找的模式后，该把它们应用到输入流上来发现可能的匹配了。为了在事件流上运行你的模式，需要创建一个 PatternStream 。 给定一个输入流 input ，一个模式 pattern 和一个可选的用来对使用事件时间时有同样时间戳或者同时到达的事件进行排序的比较器 comparator

CEP的时间处理