3. SQL 客户端与编程模型：SQL Client 模式、Table API 与 SQL 的关系、编程模型（Source-Transform-Sink）
这一章，我们来聊聊 Flink SQL 开发的几个核心概念。说白了，就是搞清楚「怎么跑 SQL」、「Table API 和 SQL 到底啥关系」以及「Flink 程序的基本骨架长什么样」。
我刚开始接触 Flink 时，也在这几个概念上绕了不少弯路。尤其是 SQL Client 和 Table API，总觉得它们功能重叠，不知道什么时候该用哪个。后来踩过坑才明白——它们其实是同一枚硬币的两面。
3.1 SQL Client 模式：交互式开发的利器
先说说 SQL Client。这是个什么东西呢？
Flink 官方提供了一个命令行工具，叫 sql-client。你可以在终端里直接输入 SQL 语句，实时看到结果。有点像 MySQL 的客户端，但背后跑的是 Flink 引擎。
核心价值：SQL Client 让你不用写一行 Java/Scala 代码，就能完成 Flink SQL 的开发、调试和运维。
我个人习惯把 SQL Client 用在两个场景：
- 快速验证：写了一段 SQL，不确定语法对不对、逻辑通不通，直接丢进 SQL Client 跑一下。
- 临时查询：线上有个问题，需要查一下 Kafka 里的数据长什么样，或者看看 Hive 表里有没有脏数据。
启动 SQL Client 很简单：
# 进入 Flink 安装目录
./bin/sql-client.sh embedded
# 或者指定配置文件
./bin/sql-client.sh embedded -d conf/sql-client.yaml进去之后，你会看到一个 Flink SQL> 的提示符。这时候就可以开始写 SQL 了。
小技巧：我建议你在 sql-client.yaml 里提前配好常用的 Catalog 和 Connector。比如 Kafka 的连接信息、Hive 的元数据地址。这样每次启动后直接就能用，不用反复写 CREATE TABLE 语句。
SQL Client 支持两种运行模式：
| 模式 | 说明 | 适用场景 |
|---|---|---|
| embedded | 客户端进程内嵌 Flink 运行时，不依赖集群 | 本地开发、小数据量验证 |
| gateway | 连接远程 Flink Session 集群 | 生产环境、多人协作 |
我记得有一次线上排查数据延迟问题，就是靠 SQL Client 的 gateway 模式连上集群，直接查 Kafka Source 的 watermark 和当前处理时间，几分钟就定位到了问题。要是没有这个工具，估计得写个 Flink 作业再提交，折腾半天。
3.2 Table API 与 SQL 的关系：一体两面
很多人觉得 Table API 和 SQL 是两套东西。其实不然。
Flink 的 Table API 和 SQL 共享同一个底层引擎——Table Planner。它们最终都会被翻译成相同的 DataStream 或 DataSet 程序执行。
你可以这样理解：
- SQL：声明式，你只管「要什么」，不用管「怎么做」。
- Table API：嵌入式 DSL，用 Java/Scala 的方法链来表达查询逻辑。
举个例子，同样一个过滤+聚合操作：
-- SQL 写法
SELECT user_id, COUNT(*) AS cnt
FROM orders
WHERE amount > 100
GROUP BY user_id;// Table API 写法
tableEnv.from("orders")
.filter($("amount").isGreater(100))
.groupBy($("user_id"))
.select($("user_id"), $("count").as("cnt"));看到没？逻辑完全一样，只是表达方式不同。
关键点：Table API 和 SQL 可以无缝混用。你可以在 Table API 中调用 tableEnv.sqlQuery() 执行 SQL，也可以在 SQL 中引用 Table API 注册的临时表。
我个人更倾向于：
- 简单的 ETL 逻辑用 SQL，可读性强，维护成本低。
- 复杂的业务逻辑（比如多步状态计算、自定义 UDF 组合）用 Table API，因为 IDE 有代码提示，编译期就能发现类型错误。
注意：Table API 和 SQL 的优化路径完全一致。不要觉得「用 SQL 性能就差，用 Table API 性能就好」。性能取决于你写的逻辑和 Planner 的优化能力，跟 API 形式无关。
3.3 编程模型：Source → Transform → Sink
Flink 的编程模型，说白了就是三个步骤：
- Source：从哪里读数据？
- Transform：怎么处理数据？
- Sink：结果写到哪里去？
这个模型贯穿 Flink 的所有 API 层面——不管是 DataStream、Table API 还是 SQL，底层都是这个骨架。
下面这张图可以帮你直观理解：
在 Flink SQL 中，这三个步骤对应着不同的 DDL 和 DML 语句：
| 阶段 | SQL 语句 | 作用 |
|---|---|---|
| Source | CREATE TABLE ... WITH ('connector'='kafka', ...) |
定义输入表，指定数据源连接信息 |
| Transform | INSERT INTO sink_table SELECT ... FROM source_table |
编写查询逻辑，处理数据 |
| Sink | CREATE TABLE ... WITH ('connector'='jdbc', ...) |
定义输出表，指定结果写入目标 |
一个完整的 Flink SQL 作业，通常长这样：
-- 1. 定义 Source（从 Kafka 读订单数据）
CREATE TABLE orders (
order_id BIGINT,
user_id BIGINT,
amount DOUBLE,
ts TIMESTAMP(3),
WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'orders',
'properties.bootstrap.servers' = 'localhost:9092',
'format' = 'json'
);
-- 2. 定义 Sink（写入 MySQL）
CREATE TABLE order_stats (
user_id BIGINT,
total_amount DOUBLE,
order_count BIGINT,
PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:mysql://localhost:3306/flink',
'table-name' = 'order_stats'
);
-- 3. Transform + Sink（执行计算并写入结果）
INSERT INTO order_stats
SELECT
user_id,
SUM(amount) AS total_amount,
COUNT(*) AS order_count
FROM orders
WHERE amount > 0
GROUP BY user_id;经验之谈：我建议你在写 Flink SQL 作业时，先把 Source 和 Sink 的表定义写好，单独测试连通性。确认数据能读进来、能写出去之后，再写中间的 Transform 逻辑。这样排查问题会快很多。
3.4 三种开发模式的对比
Flink SQL 开发有三种主流模式，我整理了一张对比表：
| 模式 | 开发方式 | 适合场景 | 优点 | 缺点 |
|---|---|---|---|---|
| SQL Client | 命令行交互 | 快速验证、临时查询 | 零编码、即时反馈 | 不适合复杂逻辑、难以复用 |
| Table API | Java/Scala 代码 | 复杂 ETL、需要 IDE 支持 | 类型安全、可调试、可复用 | 代码量较大 |
| SQL + Java 混合 | 代码中嵌入 SQL | 生产作业、需要灵活扩展 | 兼顾 SQL 简洁性和代码灵活性 | 需要同时掌握两种语法 |
我个人在实际项目中，90% 的场景用的是第三种——SQL + Java 混合模式。为什么呢？
- 核心的 ETL 逻辑用 SQL 写，清晰易懂，业务同学也能 review。
- 外围的配置管理、错误处理、监控上报用 Java 写，灵活可控。
避坑指南：我曾经在一个项目里，把所有逻辑都写在 SQL Client 的脚本里，结果作业一多，管理起来非常痛苦。后来改成 Java 项目 + SQL 文件的方式，每个作业一个独立的 SQL 文件，通过代码动态加载执行。这样既保留了 SQL 的可读性，又有了版本管理和 CI/CD 的能力。
3.5 小结
这一章我们聊了三个核心概念：
- SQL Client：交互式开发工具，适合快速验证和临时查询。
- Table API 与 SQL 的关系：同根同源，可以混用，选择取决于场景偏好。
- Source → Transform → Sink：Flink 编程模型的基本骨架，所有作业都遵循这个模式。