【大数据】Flink SQL 语法篇（二）：WITH、SELECT & WHERE、SELECT DISTINCT

本文介绍: 关于看如何看一段 Flink SQL 最终的执行计划：最好的方法就如上图，看 Flink Web UI 的算子图，算子图上详细的标记清楚了每一个算子做的事情。

应用场景（支持 Batch / Streaming）：With 语句和离线 Hive SQL With 语句一样的，语法糖 +1，使用它可以让你的代码逻辑更加清晰。

-- 语法糖 +1
WITH orders_with_total AS (
    SELECT order_id, price + tax AS total
    FROM Orders
)
SELECT order_id, SUM(total)
FROM orders_with_total
GROUP BY order_id;

应用场景（支持 Batch / Streaming）：SELECT & WHERE 语句和离线 Hive SQL 语句一样的，常用作 ETL，过滤，字段清洗标准化。

INSERT INTO target_table
SELECT * FROM Orders

INSERT INTO target_table
SELECT order_id, price + tax FROM Orders

INSERT INTO target_table
-- 自定义 Source 的数据
SELECT order_id, price FROM (VALUES (1, 2.0), (2, 3.1))  AS t (order_id, price)

INSERT INTO target_table
SELECT price + tax FROM Orders WHERE id = 10

-- 使用 UDF 做字段标准化处理
INSERT INTO target_table
SELECT PRETTY_PRINT(order_id) FROM Orders
-- 过滤条件
Where id > 3

其实理解一个 SQL 最后生成的任务是怎样执行的，最好的方式就是理解其语义。

以下面的 SQL 为例，我们来介绍下其在离线中和在实时中执行的区别，对比学习一下，大家就比较清楚了。

INSERT INTO target_table
SELECT PRETTY_PRINT(order_id) FROM Orders
Where id > 3

这个 SQL 对应的实时任务，假设 Orders 为 Kafka，target_table 也为 Kafka，在执行时，会生成三个算子：

可以看到这个实时任务的所有算子是以一种 Pipeline 模式运行的，所有的算子在同一时刻都是处于 running 状态的，24 小时一直在运行，实时任务中也没有离线中常见的分区概念。

那么如果这个 SQL 放在 Hive 中执行时，假设其中 Orders 为 Hive 表，target_table 也为 Hive 表，其也会生成三个类似的算子（虽然实际可能会被优化为一个算子，这里为了方便对比，划分为三个进行介绍），离线和实时任务的执行方式完全不同：

应用场景（支持 Batch / Streaming）：语句和离线 Hive SQL SELECT DISTINCT 语句一样的，用作根据 key 进行数据去重。

INSERT into target_table
SELECT DISTINCT id 
FROM Orders

这个 SQL 对应的实时任务，假设 Orders 为 kafka，target_table 也为 Kafka，在执行时，会生成三个算子：

在这里插入图片描述
对于实时任务，计算时的状态可能会无限增长。状态大小取决于不同 key（上述案例为 id 字段）的数量。为了防止状态无限变大，我们可以设置状态的 TTL。但是这可能会影响查询结果的正确性，比如某个 key 的数据过期从状态中删除了，那么下次再来这么一个 key，由于在状态中找不到，就又会输出一遍。