关于Flink:深入解读-Flink-SQL-113

本文由社区志愿者陈政羽整顿，Apache Flink 社区在 5 月份公布了 1.13 版本，带来了很多新的变动。文章整顿自徐榜江 (雪尽) 5 月 22 日在北京的 Flink Meetup 分享的《深刻解读 Flink SQL 1.13》，内容包含：

1. Flink SQL 1.13 概览
2. 外围 feature 解读
3. 重要改良解读
4. Flink SQL 1.14 将来布局
5. 总结

一、Flink SQL 1.13 概览

Flink 1.13 是一个社区大版本，解决的 issue 在 1000 个以上，通过上图咱们能够看到，解决的问题大部分是对于 Table/SQL 模块，一共 400 多个 issue 占了总体的 37% 左右。这些 issue 次要围绕了 5 个 FLIP 开展，在本文中咱们也会依据这 5 个方面进行介绍，它们别离是：

上面咱们对这些 FLIP 进行具体解读。

二、外围 feature 解读

1. FLIP-145：反对 Window TVF

社区的小伙伴应该理解，在腾讯、阿里巴巴、字节跳动等公司的外部分支曾经开发了这个性能的根底版本。这次 Flink 社区也在 Flink 1.13 推出了 TVF 的相干反对和优化。上面将从 Window TVF 语法、近实时累计计算场景、Window 性能优化、多维数据分析，来剖析这个新性能。

1.1 Window TVF 语法

在 1.13 版本前，window 的实现是通过一个非凡的 SqlGroupedWindowFunction：

SELECT 
    TUMBLE_START(bidtime,INTERVAL '10' MINUTE),
  TUMBLE_END(bidtime,INTERVAL '10' MINUTE),
  TUMBLE_ROWTIME(bidtime,INTERVAL '10' MINUTE),
  SUM(price)
FROM MyTable
GROUP BY TUMBLE(bidtime,INTERVAL '10' MINUTE)

在 1.13 版本中，咱们对它进行了 Table-Valued Function 的语法标准化：

SELECT WINDOW_start,WINDOW_end,WINDOW_time,SUM(price) 
FROM Table(TUMBLE(Table myTable,DESCRIPTOR(biztime),INTERVAL '10' MINUTE))
GROUP BY WINDOW_start,WINDOW_end

通过比照两种语法，咱们能够发现：TVF 语法更加灵便，不须要必须跟在 GROUP BY 关键字前面，同时 Window TVF 基于关系代数，使得其更加规范。在只须要划分窗口场景时，能够只用 TVF，无需用 GROUP BY 做聚合，这使得 TVF 扩展性和表达能力更强，反对自定义 TVF（例如实现 TOP-N 的 TVF）。

上图中的示例就是利用 TVF 做的滚动窗口的划分，只须要把数据划分到窗口，无需聚合；如果后续须要聚合，再进行 GROP BY 即可。同时，对于相熟批 SQL 的用户来说，这种操作是十分天然的，咱们不再须要像 1.13 版本之前那样必须要用非凡的 SqlGroupedWindowFunction 将窗口划分和聚合绑定在一起。

目前 Window TVF 反对 tumble window，hop window，新增了 cumulate window；session window 预计在 1.14 版本也会反对。

1.2 Cumulate Window

Cumulate window 就是累计窗口，简略来说，以上图外面时间轴上的一个区间为窗口步长。

• 第一个 window 统计的是一个区间的数据；
• 第二个 window 统计的是第一区间和第二个区间的数据；
• 第三个 window 统计的是第一区间，第二个区间和第三个区间的数据。

累积计算在业务场景中十分常见，如累积 UV 场景。在 UV 大盘曲线中：咱们每隔 10 分钟统计一次当天累积用户 UV。

在 1.13 版本之前，当须要做这种计算时，咱们个别的 SQL 写法如下：

INSERT INTO cumulative_UV
SELECT date_str,MAX(time_str),COUNT(DISTINCT user_id) as UV
FROM (
    SELECT
      DATE_FORMAT(ts,'yyyy-MM-dd') as date_str,
      SUBSTR(DATE_FORMAT(ts,'HH:mm'),1,4) || '0' as time_str,
      user_id
  FROM user_behavior
)
GROUP BY date_str

先将每条记录所属的工夫窗口字段拼接好，而后再对所有记录依照拼接好的工夫窗口字段，通过 GROUP BY 做聚合，从而达到近似累积计算的成果。

• 1.13 版本前的写法有很多毛病，首先这个聚合操作是每条记录都会计算一次。其次，在追逆数据的时候，生产沉积的数据时，UV 大盘的曲线就会跳变。
• 在 1.13 版本反对了 TVF 写法，基于 cumulate window，咱们能够批改为上面的写法，将每条数据依照 Event Time 准确地分到每个 Window 外面, 每个窗口的计算通过 watermark 触发，即便在追数据场景中也不会跳变。

INSERT INTO cumulative_UV
SELECT WINDOW_end,COUNT(DISTINCT user_id) as UV
FROM Table(CUMULATE(Table user_behavior,DESCRIPTOR(ts),INTERVAL '10' MINUTES,INTERVAL '1' DAY))
)
GROUP BY WINDOW_start,WINDOW_end

UV 大盘曲线成果如下图所示：

1.3 Window 性能优化

Flink 1.13 社区开发者们对 Window TVF 进行了一系列的性能优化，包含：

• 内存优化： 通过内存预调配，缓存 window 的数据，通过 window watermark 触发计算，通过申请一些内存 buffer 防止高频的拜访 state；
• 切片优化： 将 window 切片，尽可能复用已计算结果，如 hop window，cumulate window。计算过的分片数据无需再次计算，只需对切片的计算结果进行复用；
• 算子优化：window 算子反对 local-global 优化；同时反对 count(distinct) 主动解热点优化；
• 早退数据： 反对将早退数据计算到后续分片，保证数据准确性。

基于这些优化，咱们通过开源 Benchmark (Nexmark) 进行性能测试。结果显示 window 的普适性能有 2x 晋升，且在 count(distinct) 场景会有更好的性能晋升。

1.4 多维数据分析

语法的标准化带来了更多的灵活性和扩展性，用户能够间接在 window 窗口函数上进行多维分析。如下图所示，能够间接进行 GROUPING SETS、ROLLUP、CUBE 的剖析计算。如果是在 1.13 之前的版本，咱们可能须要对这些分组进行独自的 SQL 聚合，再对聚合后果做 union 操作能力达到相似的成果。而当初，相似这种多维分析的场景，能够间接在 window TVF 上反对。

反对 Window Top-N

除了多维分析，Window TVF 也反对 Top-N 语法，使得在 Window 上取 Top-N 的写法更加简略。

2. FLIP-162：时区和工夫函数

2.1 时区问题剖析

大家在应用 Flink SQL 时反馈了很多时区相干的问题，造成时区问题的起因能够演绎为 3 个：

• PROCTIME() 函数应该思考时区，但未思考时区；
• CURRENT_TIMESTAMP/CURRENT_TIME/CURRENT_DATE/NOW() 函数未思考时区；

• Flink 的工夫属性，只反对定义在 TIMESTAMP 这种数据类型下面，这个类型是无时区的，TIMESTAMP 类型不思考时区，但用户心愿是本地时区的工夫。

针对 TIMESTAMP 类型没有思考时区的问题，咱们提议通过 TIMESTAMP_LTZ 类型反对 (TIMESTAMP_LTZ 是 timestamp with local time zone 的缩写)。能够通过上面的表格来进行和 TIMESTAMP 的比照：

TIMESTAMP_LTZ 区别于之前咱们应用的 TIMESTAMP，它示意相对工夫的含意。通过比照咱们能够发现：

• 如果咱们配置应用 TIMESTAMP，它能够是字符串类型的。用户不论是从英国还是中国时区来察看，这个值都是一样的；
• 然而对于 TIMSTAMP_TLZ 来说，它的起源就是一个 Long 值，示意从工夫原点流逝过的工夫。同一时刻，从工夫原点流逝的工夫在所有时区都是雷同的，所以这个 Long 值是相对工夫的概念。当咱们在不同的时区去察看这个值，咱们会用本地的时区去解释成“年 - 月 - 日 - 时 - 分 - 秒”的可读格局，这就是 TIMSTAMP_TLZ 类型，TIMESTAMP_LTZ 类型也更加合乎用户在不同时区下的应用习惯。

上面的例子展现了 TIMESTAMP 和 TIMESTAMP_LTZ 两个类型的区别。

2.2 工夫函数纠正

勘误 PROCTIME() 函数

当咱们有了 TIMESTAMP_LTZ 这个类型的时候，咱们对 PROCTIME() 类型做了纠正：在 1.13 版本之前，它总是返回 UTC 的 TIMESTAMP；而当初，咱们把返回类型变为了 TIMESTAMP_LTZ。PROCTIME 除了示意函数之外，也能够示意工夫属性的标记。

勘误 CURRENT_TIMESTAMP/CURRENT_TIME/CURRENT_DATE/NOW() 函数

这些函数在不同时区下进去的值是会发生变化的。例如在英国 UTC 时区时候是凌晨 2 点；然而如果你设置了时区是 UTC+8，工夫就是在早上的 10 点。不同时区的理论工夫会发生变化，成果如下图：

解决 processing time Window 时区问题

大家都晓得 proctime 能够示意一个工夫属性，对 proctime 的 window 操作：

• 在 1.13 版本之前，如果咱们须要做按天的 window 操作，你须要手动解决时区问题，去做一些 8 小时的偏移而后再减回去；
• 在 FLIP-162 中咱们解决了这个问题，当初用户应用的时候非常简略，只须要申明 proctime 属性，因为 PROCTIME() 函数的返回值是 TIMESTAMP_LTZ，所以后果是会思考本地的时区。下图的例子显示了在不同的时区下，proctime 属性的 window 的聚合是依照本地时区进行的。

勘误 Streaming 和 Batch 模式下函数取值形式

工夫函数其实在流和批下面的表现形式会有所区别，这次修改次要是让其更加合乎用户理论的应用习惯。例如以下函数：

• 在流模式中是 per-record 计算，即每条数据都计算一次；
• 在 Batch 模式是 query-start 计算，即在作业开始前计算一次。例如咱们罕用的一些 Batch 计算引擎，如 Hive 也是在每一个批开始前计算一次。

2.3 工夫类型应用

在 1.13 版本也反对了在 TIMESTAMP 列上定义 Event time，也就是说 Event time 当初既反对定义在 TIMESTAMP 列上，也反对定义在 TIMESTAMP_ LTZ 列上。那么作为用户，具体什么场景用什么类型呢？

• 当作业的上游源数据蕴含了字符串的工夫（如：2021-4-15 14:00:00）这样的场景，间接申明为 TIMESTAMP 而后把 Event time 定义在下面即可，窗口在计算的时候会基于工夫字符串进行切分，最终会计算出合乎你理论想要的料想后果；

• 当上游数据源的打点工夫属于 long 值，示意的是一个相对工夫的含意。在 1.13 版本你能够把 Event time 定义在 TIMESTAMP_LTZ 下面。此时定义在 TIMESTAMP_LTZ 类型上的各种 WINDOW 聚合，都可能主动的解决 8 小时的时区偏移问题，无需依照之前的 SQL 写法额定做时区的批改和勘误。

小提示：Flink SQL 中对于工夫函数，时区反对的这些晋升，是版本不兼容的。用户在进行版本更新的时候须要注意作业逻辑中是否蕴含此类函数，防止降级后业务受到影响。

2.4 夏令时反对

在 Flink 1.13 以前，对于国外夏令时时区的用户，做窗口相干的计算操作是十分困难的一件事，因为存在夏令时和冬令时切换的跳变。

Flink 1.13 通过反对在 TIMESTAMP_LTZ 列上定义工夫属性，同时 Flink SQL 在 WINDOW 解决时奇妙地联合 TIMESTAMP 和 TIMESTAMP_LTZ 类型，优雅地反对了夏令时。这对国外夏令时时区用户，以及有海内业务场景的公司比拟有用。

三、重要改良解读

1. FLIP-152：晋升 Hive 语法兼容性

FLIP-152 次要是做了 Hive 语法的兼容性加强，反对了 Hive 的一些罕用 DML 和 DQL 语法，包含：

通过 Hive dialect 反对 Hive 罕用语法。Hive 有很多的内置函数，Hive dialect 须要配合 HiveCatalog 和 Hive Module 一起应用，Hive Module 提供了 Hive 所有内置函数，加载后能够间接拜访。

与此同时，咱们还能够通过 Hive dialect 创立 / 删除 Catalog 函数以及一些自定义的函数，这样使得 Flink SQL 与 Hive 的兼容性失去了极大的晋升，让相熟 Hive 的用户应用起来会更加不便。

2. FLIP-163：改良 SQL Client

在 1.13 版本之前，大家感觉 Flink SQL Client 就是周边的一个小工具。然而，FLIP-163 在 1.13 版本进行了重要改良：

1. 通过 -i 的参数，提前把 DDL 一次性加载初始化，不便初始化表的多个 DDL 语句，不须要屡次执行命令创立表，代替了之前用 yaml 文件形式创立表；
2. 反对 -f 参数，其中 SQL 文件反对 DML（insert into）语句；

3. 反对更多实用的配置：

   • 通过 SET SQL-client.verbose = true , 开启 verbose，通过开启 verbose 打印整个信息，绝对以前只输入一句话更加容易追踪错误信息；
   • 通过 SET execution.runtime-mode=streaming / batch 反对设置批 / 流作业模式；
   • 通过 SET pipline.name=my_Flink_job 设置作业名称；
   • 通过 SET execution.savepoint.path=/tmp/Flink-savepoints/savepoint-bb0dab 设置作业 savepoint 门路；
   • 对于有依赖的多个作业，通过 SET Table.dml-sync=true 去抉择是否异步执行，例如离线作业，作业 a 跑完能力跑作业 b，通过设置为 true 实现执行有依赖关系的 pipeline 调度。

4. 同时反对 STATEMENT SET 语法：

   有可能咱们的一个查问不止写到一个 sink 外面，而是须要输入到多个 sink，比方一个 sink 写到 jdbc，一个 sink 写到 HBase。

   • 在 1.13 版本之前须要启动 2 个 query 去实现这个作业；
   • 在 1.13 版本，咱们能够把这些放到一个 statement 外面，以一个作业的形式去执行，可能实现节点的复用，节约资源。

3. FLIP-136：加强 DataStream 和 Table 的转换

尽管 Flink SQL 大大降低了咱们应用实时计算的一些应用门槛，但 Table/SQL 这种高级封装也屏蔽了一些底层实现，如 timer，state 等。不少高级用户心愿可能间接操作 DataStream 取得更多的灵活性，这就须要在 Table 和 DataStream 之间进行转换。FLIP-136 加强了 Table 和 DataStream 间的转换，使得用户在两者之间的转换更加容易。

• 反对 DataStream 和 Table 转换时传递 EVENT TIME 和 WATERMARK；

Table Table = TableEnv.fromDataStream(
    dataStream,
  Schema.newBuilder()
  .columnByMetadata("rowtime","TIMESTMP(3)")
  .watermark("rowtime","SOURCE_WATERMARK()")
  .build());
)

• 反对 Changelog 数据流在 Table 和 DataStream 间互相转换。

//DATASTREAM 转 Table
StreamTableEnvironment.fromChangelogStream(DataStream<ROW>): Table
StreamTableEnvironment.fromChangelogStream(DataStream<ROW>,Schema): Table
//Table 转 DATASTREAM
StreamTableEnvironment.toChangelogStream(Table): DataStream<ROW>
StreamTableEnvironment.toChangelogStream(Table,Schema): DataStream<ROW>  

四、Flink SQL 1.14 将来布局

1.14 版本次要有以下几点布局：

• 删除 Legacy Planner：从 Flink 1.9 开始，在阿里奉献了 Blink-Planner 之后，很多一些新的 Feature 曾经基于此 Blink Planner 进行开发，以前旧的 Legacy Planner 会彻底删除；
• 欠缺 Window TVF：反对 session window，反对 window TVF 的 allow -lateness 等；
• 晋升 Schema Handling：全链路的 Schema 解决能力以及要害校验的晋升；
• 加强 Flink CDC 反对 ：加强对上游 CDC 零碎的集成能力，Flink SQL 内更多的算子反对 CDC 数据流。

五、总结

本文具体解读了 Flink SQL 1.13 的外围性能和重要改良。

• 反对 Window TVF；
• 系统地解决时区和工夫函数问题；
• 晋升 Hive 和 Flink 的兼容性；
• 改良 SQL Client；
• 加强 DataStream 和 Table 的转换。

同时还分享了社区对于 Flink SQL 1.14 的将来布局，置信看完文章的同学能够对 Flink SQL 在这个版本中的变动有更多的理解，在实际过程中大家能够多多关注这些新的改变和变动，感触它们所带来的业务层面上的便捷。