Flink 技术与应用（一）

Flink技术与应用（初级篇）

起源

Apache Flink 是一个开源的大数据处理框架，其起源可以追溯到一个名为 Stratosphere 的研究项目，旨在建立下一代大数据分析引擎，2010 年，从 Stratosphere 项目中分化出了 Flink 的前身。到了 2014 年，Flink 项目被捐赠给了 Apache 软件基金会，并在同年 4 月成为 Apache 的孵化项目，12 月成为 Apache 的顶级项目。

在德语中，Flink 一词表示快速和灵巧，项目采用一只松鼠的彩色图案作为 logo，这不仅是因为松鼠具有快速和灵巧的特点，还因为柏林的松鼠有一种迷人的红棕色，而 Flink 的松鼠 logo 拥有可爱的尾巴，尾巴的颜色与 Apache 软件基金会的 logo 颜色相呼应，也就是说，这是一只 Apache 风格的松鼠。

以下为Apache 官网对Flink的解释：

Apache Flink is a framework and distributed processing engine for stateful computations over unbounded and bounded data streams. Flink has been designed to run in all common cluster environments perform computations at in-memory speed and at any scale.

无界和有界代表Flink既可以处理离线数据，也可以处理实时数据，即流批一体。既然在此已经提到流批一体了，那么Flink 是如何同时实现批处理与流处理的呢？答案是，Flink 将批处理（即处理有限的静态数据）视作一种特殊的流处理。而在 Spark 生态体系中，对于批处理和流处理采用了不同的技术框架，批处理由 SparkSQL 实现，流处理由 Spark Streaming 实现，这也是大部分框架采用的策略，使用独立的处理器实现批处理和流处理，而 Flink 可以同时实现批处理和流处理。

上图为Flink的核心概念，在本文中我们主要尝试解释，运行架构的部分内容、窗口函数以及SQL关联三部分内容。

架构与原理

JobManager 负责协调和管理整个作业的执行

TaskManager 负责执行具体的任务，并进行数据的缓存和交换

Flink 程序结构

Flink 程序的基本构建块是流和转换。从概念上讲，流是（可能永无止境的）数据记录流，而转换是将一个或多个流作为一个或多个流的操作输入，并产生一个或多个输出流。

Flink 应用程序结构就是如上图所示：

Source: 数据源，Flink 在流处理和批处理上的 source 大概有 4 类：基于本地集合的 source、基于文件的 source、基于网络套接字的 source、自定义的 source。自定义的 source 常见的有 Apache kafka、RabbitMQ 等，当然你也可以定义自己的 source。

Transformation：数据转换的各种操作，有 Map / FlatMap / Filter / KeyBy / Reduce / Fold / Aggregations / Window / WindowAll / Union / Window join / Split / Select等，操作很多，可以将数据转换计算成你想要的数据。

Sink：接收器，Flink 将转换计算后的数据发送的地点，你可能需要存储下来，Flink 常见的 Sink 大概有如下几类：写入文件、打印出来、写入 socket 、自定义的 sink 。自定义的 sink 常见的有 Apache kafka、RabbitMQ、MySQL、ElasticSearch、Apache Cassandra、Hadoop FileSystem 等，同理你也可以定义自己的 sink。

由于我周边的同事大部分都是做数据模型和数据分析的，因此上面的内容只需要了解就可以，并不需要完全的掌握；对于做系统或工程的同事，如果有更高的追求成为Flink的commiter ，这些内容就有太少了，主要研究源码。

并行度

当要处理的数据量非常大时，我们可以把一个算子操作，“复制”多份到多个节点，数据来了之后就可以到其中任意一个执行。这样一来，一个算子任务就被拆分成了多个并行的“子任务”（subtask），再将它们分发到不同节点，就真正实现了并行计算。一个特定算子的子任务的个数被称之为其并行度（parallelism），一个流程序的并行度，可以认为就是其所有算子中最大的并行度，而一个程序中，不同的算子可能具有不同的并行度。

行度的设置（优先级从高到低由上至下）：

算子设置stream.map(word -> Tuple2.of(word, 1L)).setParallelism(2)；

环境设置env.setParallelism(2)；

提交设置bin/flink run –p 2 –c ......

配置文件设置parallelism.default: 2

初始值为1，在没有指定并行度的时候，就会采用配置文件中的集群默认并行度，在开发环境中，没有配置文件，默认并行度就是当前机器的CPU核心数

Tasks 和算子链

一个数据流在算子之间传输数据的形式可以是一对一（one-to-one）的直通（forwarding）模式，也可以是打乱的重分区（redistributing）模式。

一对一：不需要重新分区，也不需要调整数据的顺序，这种关系类似于Spark中的窄依赖
重分区：每一个算子的子任务，会根据数据传输的策略，把数据发送到不同的下游目标任务，这些传输方式都会引起重分区的过程，这一过程类似于Spark中的shuffle，执行图中只要不是forwarding，就一定是重分区方式

合并算子链：并行度相同的一对一算子操作，可以直接链接在一起形成一个“大”的任务（task），这样合并的task会被一个线程执行，这样就可以减少：

线程之间的切换

消息的序列化/反序列化

基于缓存区的数据交换

在减少时延的同时提升吞吐量

下图中样例数据流用 5 个 subtask 执行，因此有 5 个并行线程。

Task Slots 和资源

每个 worker（TaskManager）都是一个 JVM 进程，可以在单独的线程中执行一个或多个 subtask。为了控制一个 TaskManager 中接受多少个 task，就有了所谓的 task slots（至少一个）。

每个 task slot 代表 TaskManager 中资源的固定子集。例如，具有 3 个 slot 的 TaskManager，会将其托管内存 1/3 用于每个 slot。分配资源意味着 subtask 不会与其他作业的 subtask 竞争托管内存，而是具有一定数量的保留托管内存。注意此处没有 CPU 隔离；当前 slot 仅分离 task 的托管内存。

通过调整 task slot 的数量，用户可以定义 subtask 如何互相隔离。每个 TaskManager 有一个 slot，这意味着每个 task 组都在单独的 JVM 中运行（例如，可以在单独的容器中启动）。具有多个 slot 意味着更多 subtask 共享同一 JVM。同一 JVM 中的 task 共享 TCP 连接（通过多路复用）和心跳信息。它们还可以共享数据集和数据结构，从而减少了每个 task 的开销。

默认情况下，Flink 允许 subtask 共享 slot，即便它们是不同的 task 的 subtask，只要是来自于同一作业即可。结果就是一个 slot 可以持有整个作业管道。允许 slot 共享有两个主要优点：

Flink 集群所需的 task slot 和作业中使用的最大并行度恰好一样。无需计算程序总共包含多少个 task（具有不同并行度）。
容易获得更好的资源利用。如果没有 slot 共享，非密集 subtask（source/map()）将阻塞和密集型 subtask（window）一样多的资源。通过 slot 共享，我们示例中的基本并行度从 2 增加到 6，可以充分利用分配的资源，同时确保繁重的 subtask 在 TaskManager 之间公平分配。

回顾数据湖配置参数

关系：任务槽 & 并行度

任务槽的数量决定了 Flink 集群可以同时执行的任务数量。如果一个 TaskManager 有 n 个任务槽，那么它可以同时执行 n 个任务或子任务。并行度则决定了每个算子可以被拆分成多少个子任务来执行。如果一个算子的并行度设置为 m，那么它将被拆分成 m 个子任务，这些子任务需要 m 个任务槽来执行。

由此可知，上图中的并行度只要<= 4* 4 就没问题。

流式分析

Flink 是流式的、实时的计算引擎。

流式：就是数据源源不断的流进来，也就是数据没有边界，但是我们计算的时候必须在一个有边界的范围内进行，所以这里面就有一个问题，边界怎么确定？无非就两种方式，根据时间段或者数据量进行确定，根据时间段就是每隔多长时间就划分一个边界，根据数据量就是每来多少条数据划分一个边界，Flink 中就是这么划分边界的，本文会详细讲解。

实时：就是数据发送过来之后立马就进行相关的计算，然后将结果输出。这里的计算有两种：

一种是只有边界内的数据进行计算，这种好理解，比如统计每个用户最近五分钟内浏览的新闻数量，就可以取最近五分钟内的所有数据，然后根据每个用户分组，统计新闻的总数。

另一种是边界内数据与外部数据进行关联计算，比如：统计最近五分钟内浏览新闻的用户都是来自哪些地区，这种就需要将五分钟内浏览新闻的用户信息与 hive 中的地区维表进行关联，然后在进行相关计算。

Flink 明确支持以下三种时间语义:

事件时间(event time)：事件产生的时间，记录的是设备生产(或者存储)事件的时间
摄取时间(ingestion time)： Flink 读取事件时记录的时间
处理时间(processing time)： Flink pipeline 中具体算子处理事件的时间

Event Time

在Flink的流式处理中，绝大部分的业务都会使用eventTime，一般只在eventTime无法使用时，才会被迫使用ProcessingTime或者IngestionTime。

如果想要使用事件时间，需要额外给 Flink 提供一个时间戳提取器和 Watermark 生成器，Flink 将使用它们来跟踪事件时间的进度。

Watermark

流处理从事件产生，到流经 source，再到 operator，中间是有一个过程和时间的，虽然大部分情况下，流到 operator 的数据都是按照事件产生的时间顺序来的，但是也不排除由于网络、背压等原因，导致乱序的产生，所谓乱序，就是指 Flink 接收到的事件的先后顺序不是严格按照事件的 Event Time 顺序排列的，所以 Flink 最初设计的时候，就考虑到了网络延迟，网络乱序等问题，所以提出了一个抽象概念：水印（WaterMark），一旦出现乱序，如果只根据 EventTime 决定 Window 的运行，我们不能明确数据是否全部到位，但又不能无限期的等下去，此时必须要有个机制来保证一个特定的时间后，必须触发 Window 去进行计算了，这个特别的机制，就是 Watermark。

Apache Flink 中的水位线（Watermark）是一种逻辑时钟机制，用于处理基于事件时间的流数据中的乱序问题，确保在事件时间窗口内正确地收集和处理数据。

Watermark = 数据的事件时间 – 最大允许的乱序时间

Watermark是一个时间戳
Watermark水位线会一直上升（变大），不会下降

窗口计算的触发时间：

窗口中有数据
Watermark >=窗口时间

In order

Out of order

Parallel Streams

Allowed Lateness

在使用 event-time 窗口时，数据可能会迟到，即 Flink 用来追踪 event-time 进展的 watermark 已经越过了窗口结束的 timestamp 后，数据才到达。

默认情况下，watermark 一旦越过窗口结束的 timestamp，迟到的数据就会被直接丢弃。但是 Flink 允许指定窗口算子最大的 allowed lateness。 Allowed lateness 定义了一个元素可以在迟到多长时间的情况下不被丢弃，这个参数默认是 0。

所以延迟是相对于 watermarks 定义的。Watermark(t) 表示事件流的时间已经到达了 t; watermark 之后的时间戳 ≤ t 的任何事件都被称之为延迟事件。

延迟案例：

在当前窗口【假设窗口范围为10-15】已经计算之后，又来了一个属于该窗口的数据【假设事件时间为13】，这时候仍会触发 Window 操作，这种数据就称为延迟数据。

那么问题来了，延迟时间怎么计算呢？

假设窗口范围为10-15，延迟时间为2s，则只要 WaterMark<15+2，并且属于该窗口，就能触发 Window 操作。而如果来了一条数据使得 WaterMark>=15+2，10-15这个窗口就不能再触发 Window 操作，即使新来的数据的 Event Time 属于这个窗口时间内。

Flink SQL

Flink SQL 是 Flink 实时计算为简化计算模型，降低用户使用实时计算门槛而设计的一套符合标准 SQL 语义的开发语言。自 2015 年开始，阿里巴巴开始调研开源流计算引擎，最终决定基于 Flink 打造新一代计算引擎，针对 Flink 存在的不足进行优化和改进，并且在 2019 年初将最终代码开源，也就是我们熟知的 Blink。Blink 在原来的 Flink 基础上最显著的一个贡献就是 Flink SQL 的实现。

Windows

窗口（Window）是处理无界流的关键所在。窗口可以将数据流装入大小有限的“桶”中，再对每个“桶”加以处理。

窗口的生命周期：

一个窗口在第一个属于它的元素到达时就会被创建，然后在时间（event 或 processing time）超过窗口的“结束时间戳 + 用户定义的 allowed lateness 时被完全删除。例如，对于一个基于 event time 且范围互不重合（滚动）的窗口策略，如果窗口设置的时长为五分钟、可容忍的迟到时间（allowed lateness）为 1 分钟，那么第一个元素落入 12:00 至 12:05 这个区间时，Flink 就会为这个区间创建一个新的窗口。当 watermark 越过 12:06 时，这个窗口将被摧毁。

滚动窗口（Tumbling Windows）

滚动窗口将每个元素分发到指定大小的窗口。滚动窗口的大小是固定的，且各自范围之间不重叠。比如说，如果你指定了滚动窗口的大小为 5 分钟，那么每 5 分钟就会有一个窗口被计算，且一个新的窗口被创建。

insert into sink_table

select

dim,

count(*) as pv,

sum(price) as sum_price,

max(price) as max_price,

min(price) as min_price,

-- 计算 uv 数

count(distinct user_id) as uv,

UNIX_TIMESTAMP(CAST(tumble_start(row_time, interval '1' minute) AS STRING)) * 1000 as window_start

from source_table

group by

dim,

tumble(row_time, interval '1' minute);

第一个参数为 事件时间的时间戳；第二个参数为 滚动窗口大小。

场景：

大促实时数据按分钟的天累计用户、金额、订单量等

另外一种写法：

Windowing table-valued functions (Windowing TVFs)窗口表值函数

INSERT INTO print_x

SELECT

window_start,

window_end,

COUNT(DISTINCT account_id) AS order_cnt,

SUM(amount) AS amount

FROM

TABLE( TUMBLE(TABLE data_gen, DESCRIPTOR(transaction_time), INTERVAL '1' MINUTES))

GROUP BY

window_start,

window_end

三个必填的参数：

TUMBLE(TABLE data, DESCRIPTOR(timecol), size [, offset ])

data ：拥有时间属性列的表。
timecol ：列描述符，决定数据的哪个时间属性列应该映射到窗口。
size ：窗口的大小（时长）。
offset ：窗口的偏移量 [非必填]。

正常情况下如果每个小时看数，譬如区间是9点-10点，但是我也可能大促是从9:30 开始的，那按照9:30-10:30 的间隔看，offset 就相当于往后推迟了30分钟。

注意：窗口偏移只影响窗口的分配，并不会影响 Watermark *

说明：

Apache Flink 提供 4 个内置的窗口表值函数：TUMBLE，HOP，CUMULATE 和 SESSION。窗口表值函数 的返回值包括原生列和附加的三个指定窗口的列，分别是：“window_start”，“window_end”，“window_time”。在流计算模式，window_time 是 TIMESTAMP 或者 TIMESTAMP_LTZ 类型（具体哪种类型取决于输入的时间字段类型）的字段。 window_time 字段用于后续基于时间的操作。

滑动窗口（Sliding Windows）

与滚动窗口类似，滑动窗口将每个元素分发到指定大小的窗口，窗口大小通过 window size 参数设置。滑动窗口需要一个额外的滑动距离（window slide）参数来控制生成新窗口的频率。因此，如果 slide 小于窗口大小，滑动窗口可以允许窗口重叠。这种情况下，一个元素可能会被分发到多个窗口。

比如说，你设置了大小为 10 分钟，滑动距离 5 分钟的窗口，你会在每 5 分钟得到一个新的窗口，里面包含之前 10 分钟到达的数据

insert into sink_table

SELECT dim,

UNIX_TIMESTAMP(CAST(hop_start(row_time, interval '1' minute, interval '5' minute) AS STRING)) * 1000 as window_start,

count(distinct user_id) as uv

FROM source_table

GROUP BY

dim,

hop(row_time, interval '1' minute, interval '5' minute);

第一个参数为事件时间的时间戳。第二个参数为滑动窗口的滑动步长。第三个参数为滑动窗口大小

另外一种写法：

Windowing table-valued functions (Windowing TVFs)窗口表值函数

SELECT window_start, window_end, count(distinct user_id) as uv

FROM TABLE(

HOP(TABLE source_table, DESCRIPTOR(row_time), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES))

GROUP BY window_start, window_end;

HOP 有四个必填参数和一个可选参数：

HOP(TABLE data, DESCRIPTOR(timecol), slide, size [, offset ])

data：拥有时间属性列的表。
timecol：列描述符，决定数据的哪个时间属性列应该映射到窗口。
slide：窗口的滑动步长。
size：窗口的大小(时长)。
offset：窗口的偏移量 [非必填]。

会话窗口（Session Windows）

会话窗口会把数据按活跃的会话分组。与滚动窗口和滑动窗口不同，会话窗口不会相互重叠，且没有固定的开始或结束时间。会话窗口在一段时间没有收到数据之后会关闭，即在一段不活跃的间隔之后。会话窗口的 assigner 可以设置固定的会话间隔（session gap）或用 session gap extractor 函数来动态地定义多长时间算作不活跃。当超出了不活跃的时间段，当前的会话就会关闭，并且将接下来的数据分发到新的会话窗口。

insert into sink_table

SELECT

dim,

UNIX_TIMESTAMP(CAST(session_start(row_time, interval '5' minute) AS STRING)) * 1000 as window_start,

count(1) as pv

FROM source_table

GROUP BY

dim,

session(row_time, interval '5' minute);

第一个参数为 事件时间的时间戳；第二个参数为 Session Gap 间隔。

另外一种写法：

Windowing table-valued functions (Windowing TVFs)窗口表值函数

SELECT window_start, window_end, item, SUM(price) AS total_price

FROM TABLE(

SESSION(TABLE Bid PARTITION BY item, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES))

GROUP BY item, window_start, window_end;

SESSION 有三个必填参数和一个可选参数：

SESSION(TABLE data [PARTITION BY(keycols, ...)], DESCRIPTOR(timecol), gap)

data：拥有时间属性列的表。
keycols：列描述符，决定会话窗口应该使用哪些列来分区数据。
timecol：列描述符，决定数据的哪个时间属性列应该映射到窗口。
gap：两个事件被认为属于同一个会话窗口的最大时间间隔。

全局窗口（Global Windows）

全局窗口将拥有相同 key 的所有数据分发到一个全局窗口。这样的窗口模式仅在你指定了自定义的 trigger 时有用。否则，计算不会发生，因为全局窗口没有天然的终点去触发其中积累的数据。

渐进式窗口（CUMULATE）

渐进式（累积）窗口是固定窗口间隔内提前触发的的滚动窗口，其实就是 Tumble Window + early-fire 的一个事件时间的版本。例如，从每日零点到当前这一分钟绘制累积 UV，其中 10:00 时的 UV 表示从 00:00 到 10:00 的 UV 总数。渐进式窗口可以认为是首先开一个最大窗口大小的滚动窗口，然后根据用户设置的触发的时间间隔将这个滚动窗口拆分为多个窗口，这些窗口具有相同的窗口起点和不同的窗口终点。

insert into sink_table

SELECT

UNIX_TIMESTAMP(CAST(window_end AS STRING)) * 1000 as window_end,

window_start,

sum(money) as sum_money,

count(distinct id) as count_distinct_id

FROM TABLE(CUMULATE(TABLE source_table, DESCRIPTOR(row_time), INTERVAL '60' SECOND, INTERVAL '1' DAY))

GROUP BY

window_start,

window_end

CUMULATE 有四个必填参数和一个可选参数：

CUMULATE(TABLE data, DESCRIPTOR(timecol), step, size)

data：拥有时间属性列的表。
timecol：列描述符，决定数据的哪个时间属性列应该映射到窗口。
step：指定连续的累积窗口之间增加的窗口大小。
size：指定累积窗口的最大宽度的窗口时间。size必须是step的整数倍。
offset：窗口的偏移量 [非必填]。

EMIT语法

双流join

Flink SQL支持对动态表进行复杂而灵活的连接操作顾名思义，就是两个数据流之间的关联。但是流的关联与SPARK SQL的批离线关联存在较大的不同。

批处理重点关注资源是否满足，数据是否存在倾斜，当然流也需要关注，下面我们从流关联独有特性逐步分析。

默认情况下，joins 的顺序是没有优化的。表的 join 顺序是在 FROM 从句指定的。可以通过把更新频率最低的表放在第一个、频率最高的放在最后这种方式来微调 join 查询的性能。

Join 类型

数据保存

不论是INNER JOIN还是OUTER JOIN 都需要对左右两边的流的数据进行保存，JOIN算子会开辟左右两个State进行数据存储，左右两边的数据到来时候，进行如下操作：

LeftEvent到来存储到LState，RightEvent到来的时候存储到RState；
LeftEvent会去RightState进行JOIN，并发出所有JOIN之后的Event到下游；
RightEvent会去LeftState进行JOIN，并发出所有JOIN之后的Event到下游。

数据Shuffle

基于on 条件进行partition，确保两个流相同的联接key会在同一个节点处理，这一点与离线中表的关联是一致。

TTL机制

Flink SQL中可以使用TTL（Time To Live）来设置数据的过期时间，以控制数据在内存或状态中的存留时间。通过设置TTL，可以自动删除过期的数据，从而节省资源并提高性能。

要在Flink SQL中设置TTL，可以使用CREATE TABLE语句的WITH选项来指定TTL的配置。以下是一个示例：

CREATE TABLE myTable (

id INT,

name STRING,

eventTime TIMESTAMP(3),

WATERMARK FOR eventTime AS eventTime - INTERVAL '5' MINUTE -- 定义Watermark

) WITH (

'connector' = 'kafka',

'topic' = 'myTopic',

'properties.bootstrap.servers' = 'localhost:9092',

'format' = 'json',

'json.fail-on-missing-field' = 'false',

'json.ignore-parse-errors' = 'true',

'ttl' = '10m' -- 设置TTL为10分钟

);

通过在CREATE TABLE语句的WITH子句中的’ttl’选项中指定TTL的值（10m），即设置数据在内存中的存活时间为10分钟。过期的数据会自动被删除。

需要注意的是，引入TTL机制会增加一定的性能和资源开销。因此，在使用TTL时需要权衡好过期时间和系统的性能需求。

CREATE TABLE myTable (

id INT,

name STRING,

event_time TIMESTAMP(3),

WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND,

PRIMARY KEY (id) NOT ENFORCED,

TTL (event_time) AS event_time + INTERVAL '1' HOUR

) WITH (

'connector.type' = 'kafka',

...

)

注意：并非所有的连接器都支持 TTL 功能，你需要检查你所使用的连接器文档来确认是否支持 TTL。以上案例没有在本地执行确认。

Regular join

Regular join 是最通用的 join 类型。在这种 join 下，join 两侧表的任何新记录或变更都是可见的，并会影响整个 join 的结果。例如：如果左边有一条新纪录，在 Product.id 相等的情况下，它将和右边表的之前和之后的所有记录进行 join。

对于流式查询，regular join 的语法是最灵活的，允许任何类型的更新（插入、更新、删除）输入表。然而，这种操作具有重要的操作意义：Flink 需要将 Join 输入的两边数据永远保持在状态中。因此，计算查询结果所需的状态可能会无限增长，这取决于所有输入表的输入数据量。你可以提供一个合适的状态 time-to-live (TTL) 配置来防止状态过大。注意：这样做可能会影响查询的正确性

INNER JOIN

根据 join 限制条件返回一个简单的笛卡尔积。目前只支持 equi-joins，即：至少有一个等值条件。不支持任意的 cross join 和 theta join。(cross join 指的是类似 SELECT * FROM table_a CROSS JOIN table_b，theta join 指的是类似 SELECT * FROM table_a, table_b）

INNER JOIN只有符合JOIN条件时候才会有JOIN结果流出到下游，比如右边最先来的1，2，3个事件，流入时候没有任何输出，因为左边还没有可以JOIN的事件；
INNER JOIN两边的数据不论如何乱序，都能够保证和传统数据库语义一致，因为我们保存了左右两个流的所有事件到state中。

OUTER JOIN

返回所有符合条件的笛卡尔积（即：所有通过 join 条件连接的行），加上所有外表没有匹配到的行。Flink 支持 LEFT、RIGHT 和 FULL outer joins。目前只支持 equi-joins，即：至少有一个等值条件。不支持任意的 cross join 和 theta join。

左流的事件当右边没有JOIN的事件时候，将右边事件列补NULL后流向下游；* 当右边事件流入发现左边已经有可以JOIN的key的时候，并且是第一个可以JOIN上的右边事件需要撤回左边下发的NULL记录，并下发JOIN完整（带有右边事件列）的事件到下游。
在Apache Flink系统内部事件类型分为正向事件标记为“+”和撤回事件标记为“-”。

时间区间Join(Interval Join)

返回一个符合 join 条件和时间限制的简单笛卡尔积。Interval join 需要至少一个 equi-join 条件和一个 join 两边都包含的时间限定 join 条件。范围判断可以定义成就像一个条件（<, <=, >=, >），也可以是一个 BETWEEN 条件，或者两边表的一个相同类型（即：处理时间或事件时间）的时间属性的等式判断。

SELECT *

FROM Orders o, Shipments s

WHERE o.id = s.order_id

AND o.order_time BETWEEN s.ship_time - INTERVAL '4' HOUR AND s.ship_time

一些有效的 interval join 时间条件：

ltime = rtime
ltime >= rtime AND ltime < rtime + INTERVAL '10' MINUTE
ltime BETWEEN rtime - INTERVAL '10' SECOND AND rtime + INTERVAL '5' SECOND

对于流式查询，对比 regular join，interval join 只支持有时间属性的非更新表。由于时间属性是递增的，Flink 从状态中移除旧值也不会影响结果的正确性。

时态/快照（Temporal Join）

在离线数仓中有快照表、增量表、拉链表，像商品数据、用户数据等类型的数据在关联的时候可能用到的一个全量，甚至要求有数据的回溯，譬如监管上报要求的是订单时间的当时用户状态属性。在flink 里面也有快照表/版本的概念，但是理解起来可能还是抽象。

时态表（Temporal table）是一个随时间变化的表：在 Flink 中被称为动态表。时态表中的行与一个或多个时间段相关联，所有 Flink 中的表都是时态的（Temporal）。时态表包含一个或多个版本的表快照，它可以是一个变化的历史表，跟踪变化（例如，数据库变化日志，包含所有快照）或一个变化的维度表，也可以是一个将变更物化的维表（例如，存放最终快照的数据表）。

核心要点：

一条实时流（动态表），关联另外一条实时流（版本表）
在进行快照挂链时，只有左表中的数据出发join动作
关联动作发生后，缓存中只保留元素最新版本数据，过期版本被移除
快照关联仅支持内关联和左关联
核心语法：FOR SYSTEM_TIME AS OF A.time_field
左右两个表都需要定义时间字段，支持事件时间和处理时间，如果是事件时间还需要定义watermark，作用是触发输出结果，如果是处理时间则不需要设置watermark。
版本表还需要定义主键，使用主键进行关联

两个普通仅追加表如何实现版本表动作：

通过定义版本表视图实现（对右表重写）：

本质是通过分组排序取最新数据，默认指定partition by 为主键，order by 为时间字段

与离线的快照表实现思路就是一致的。

事件时间 Temporal Join

基于事件时间的 Temporal join 允许对版本表进行 join。这意味着一个表可以使用变化的元数据来丰富，并在某个时间点检索其具体值。

Temporal Joins 使用任意表（左侧输入/探测端）的每一行与版本表中对应的行进行关联（右侧输入/构建端）。 Flink 使用 SQL:2011标准 中的 FOR SYSTEM_TIME AS OF 语法去执行操作。 Temporal join 的语法如下：

SELECT [column_list]

FROM table1 [AS <alias1>]

[LEFT] JOIN table2 FOR SYSTEM_TIME AS OF table1.{ proctime | rowtime } [AS <alias2>]

ON table1.column-name1 = table2.column-name1

有了事件时间属性（即：rowtime 属性），就能检索到过去某个时间点的值。这允许在一个共同的时间点上连接这两个表。版本表将存储自最后一个 watermark 以来的所有版本（按时间标识）。

例如，假设我们有一个订单表，每个订单都有不同货币的价格。为了正确地将该表统一为单一货币（如美元）,每个订单都需要与下单时相应的汇率相关联。

-- Create a table of orders. This is a standard

-- append-only dynamic table.

CREATE TABLE orders (

order_id STRING,

price DECIMAL(32,2),

currency STRING,

order_time TIMESTAMP(3),

WATERMARK FOR order_time AS order_time - INTERVAL '15' SECOND

) WITH (/* ... */);

-- Define a versioned table of currency rates.

-- This could be from a change-data-capture

-- such as Debezium, a compacted Kafka topic, or any other

-- way of defining a versioned table.

CREATE TABLE currency_rates (

currency STRING,

conversion_rate DECIMAL(32, 2),

update_time TIMESTAMP(3) METADATA FROM `values.source.timestamp` VIRTUAL,

WATERMARK FOR update_time AS update_time - INTERVAL '15' SECOND,

PRIMARY KEY(currency) NOT ENFORCED

) WITH (

'connector' = 'kafka',

'value.format' = 'debezium-json',

/* ... */

);

SELECT

order_id,

price,

orders.currency,

conversion_rate,

order_time

FROM orders

LEFT JOIN currency_rates FOR SYSTEM_TIME AS OF orders.order_time

ON orders.currency = currency_rates.currency;

order_id price currency conversion_rate order_time

======== ===== ======== =============== =========

o_001 11.11 EUR 1.14 12:00:00

o_002 12.51 EUR 1.10 12:06:00

注意1： 事件时间 temporal join 是通过左和右两侧的 watermark 触发的；这里的 INTERVAL 时间减法用于等待后续事件，以确保 join 满足预期。请确保 join 两边设置了正确的 watermark 。

注意2： 事件时间 temporal join 需要包含主键相等的条件，即：currency_rates 表的主键 currency_rates.currency 包含在条件 orders.currency = currency_rates.currency 中。

与 regular joins 相比，就算 build side（例子中的 currency_rates 表）发生变更了，之前的 temporal table 的结果也不会被影响。与 interval joins 对比，temporal join没有定义join的时间窗口。 Probe side （例子中的 orders 表）的记录总是在 time 属性指定的时间与 build side 的版本行进行连接。因此，build side 表的行可能已经过时了。随着时间的推移，不再被需要的记录版本（对于给定的主键）将从状态中删除。

处理时间 Temporal Join

基于处理时间的 temporal join 使用处理时间属性将数据与外部版本表（例如 mysql、hbase）的最新版本相关联。

通过定义一个处理时间属性，这个 join 总是返回最新的值。可以将 build side 中被查找的表想象成一个存储所有记录简单的 HashMap<K,V>。这种 join 的强大之处在于，当无法在 Flink 中将表具体化为动态表时，它允许 Flink 直接针对外部系统工作。

下面这个处理时间 temporal join 示例展示了一个追加表 orders 与 LatestRates 表进行 join。 LatestRates 是一个最新汇率的维表，比如 HBase 表，在 10:15，10:30，10:52这些时间，LatestRates 表的数据看起来是这样的：

10:15> SELECT * FROM LatestRates;

currency rate

======== ======

US Dollar 102

Euro 114

Yen 1

10:30> SELECT * FROM LatestRates;

currency rate

======== ======

US Dollar 102

Euro 114

Yen 1

10:52> SELECT * FROM LatestRates;

currency rate

======== ======

US Dollar 102

Euro 116 <==== changed from 114 to 116

Yen 1

LastestRates 表的数据在 10:15 和 10:30 是相同的。欧元（Euro）的汇率（rate）在 10:52 从 114 变更为 116。

Orders 表示支付金额的 amount 和currency的追加表。例如：在 10:15 ，有一个金额为 2 Euro 的 order。

SELECT * FROM Orders;

amount currency

====== =========

2 Euro <== arrived at time 10:15

1 US Dollar <== arrived at time 10:30

2 Euro <== arrived at time 10:52

给出下面这些表，我们希望所有 Orders 表的记录转换为一个统一的货币。

amount currency rate amount*rate

====== ========= ======= ============

2 Euro 114 228 <== arrived at time 10:15

1 US Dollar 102 102 <== arrived at time 10:30

2 Euro 116 232 <== arrived at time 10:52

目前，temporal join 还不支持与任意 view/table 的最新版本 join 时使用 FOR SYSTEM_TIME AS OF 语法。可以像下面这样使用 temporal table function 语法来实现（时态表函数）:

SELECT

o_amount, r_rate

FROM

Orders,

LATERAL TABLE (Rates(o_proctime))

WHERE

r_currency = o_currency

注意 Temporal join 不支持与 table/view 的最新版本进行 join 时使用 FOR SYSTEM_TIME AS OF 语法是出于语义考虑，因为左流的连接处理不会等待 temporal table 的完整快照，这可能会误导生产环境中的用户。处理时间 temporal join 使用 temporal table function 也存在相同的语义问题，但它已经存在了很长时间，因此我们从兼容性的角度支持它。

processing-time 的结果是不确定的。 processing-time temporal join 常常用在使用外部系统来丰富流的数据。（例如维表）

与 regular joins 的差异，就算 build side（例子中的 currency_rates 表）发生变更了，之前的 temporal table 结果也不会被影响。与 interval joins 的差异，temporal join 没有定义数据连接的时间窗口。即：旧数据没存储在状态中。

Temporal Table Function

使用 temporal table function 去 join 表的语法和 Table Function 相同。

注意：目前只支持 inner join 和 left outer join。

假设Rates是一个 temporal table function，这个 join 在 SQL 中可以被表达为：

SELECT

o_amount, r_rate

FROM

Orders,

LATERAL TABLE (Rates(o_proctime))

WHERE

r_currency = o_currency

上述 temporal table DDL 和 temporal table function 的主要区别在于：

SQL 中可以定义 temporal table DDL，但不能定义 temporal table 函数;
temporal table DDL 和 temporal table function 都支持 temporal join 版本表，但只有 temporal table function 可以 temporal join 任何表/视图的最新版本（即"处理时间 Temporal Join"）。

Lookup 维表

lookup join 通常用于使用从外部系统查询的数据来丰富表。join 要求一个表具有处理时间属性，另一个表由查找源连接器（lookup source connnector）支持。

lookup join 和上面的处理时间 Temporal Join 语法相同，右表使用查找源连接器支持。

CREATE TEMPORARY TABLE Customers (

id INT,

name STRING,

country STRING,

zip STRING

) WITH (

'connector' = 'jdbc',

'url' = 'jdbc:mysql://mysqlhost:3306/customerdb',

'table-name' = 'customers'

);

SELECT o.order_id, o.total, c.country, c.zip

FROM Orders AS o

JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c

ON o.customer_id = c.id;

Flink SQL 支持 LEFT JOIN 和 INNER JOIN 的维表关联。如上语法所示的，维表 JOIN 语法与传统的 JOIN 语法并无二异。只是 Customers维表后面需要跟上 FOR SYSTEM_TIME AS OF PROCTIME() 的关键字，其含义是每条到达的数据所关联上的是到达时刻的维表快照，也就是说，当数据到达时，我们会根据数据上的 key 去查询远程数据库，拿到匹配的结果后关联输出。这里的 PROCTIME 即 processing time。使用 JOIN 当前维表功能需要注意的是，如果维表插入了一条数据能匹配上之前左表的数据时，JOIN的结果流，不会发出更新的数据以弥补之前的未匹配。JOIN行为只发生在处理时间（processing time），即使维表中的数据都被删了，之前JOIN流已经发出的关联上的数据也不会被撤回或改变。

总结下核心特性：