概览

第一篇我们解决了"怎么写"——一条 flink run 跑起完整的 Multi-Statement SQL 脚本。这一篇解决"怎么管"：让 Flink SQL 作业的研发流程具备和 Java 后端同样的工程能力——可检测、可追溯、可回滚、自动化。

本文将深入 Flink SQL Validate 的底层原理（Calcite 解析、验证），这也是各个大厂内部实时研发流程中最基础、最重要的一环，让你不仅知道怎么用，更理解为什么它能在不连 Flink 集群的情况下精确校验语法。

像上一篇 Flink 实时数仓开发实战：像 Hive 那样用 Flink SQL 一样，本文也提供了一个示例项目 Flink SQL Bootstrap Examples - CI/CD 帮助大家快速搭建本地环境，并一步步演示 CI/CD 流水线。

为什么需要 CI/CD

Flink SQL 及 CI/CD 的引入可以从以下四个方面大大提升研发效率：

1. 代码量
2. 技术栈门槛
3. 维护成本
4. 迭代效率

这也是为什么大厂普遍使用 Flink SQL 作为实时研发的核心原因。并且 CI/CD 它还保障了四个核心能力：

能力	说明
可检测	编译不过不能合、单测不过不能上线——机器过滤低级错误，人专注逻辑
可追溯	谁改的、什么时候改的、为什么改——git log + pipeline 记录一条链路
可回滚	出了事不至于手忙脚乱，重新部署上一个版本，几分钟恢复
自动化	从提交到上线，不需要人做机器能做的事

这些能力虽然在服务端已经是家常便饭，但是对于数据开发这些能力只有大厂内部高度集成的研发平台才能提供。但这些能力往往是最基础、最核心的研发流程，它的缺失就像一个雷：不知道什么时候、在哪里就会炸一下。

接下来我们将以 flink-sql-bootstrap + Gitlab 为样板提供一种轻量、可靠的方案： 基于公司现有的服务端 CI/CD 流程（可能是 Gitlab Runner 或 Jenkins）搭建大数据自己的实时研发 CI/CD 流程。

快速开始

示例中将非常多繁琐的工作都帮读者做好了，一个命令就能够完成所有环境的安装、配置，甚至 Gitlab ssh 的创建和配置：

cd example-cicd/docker
bash setup.sh

这个脚本自动完成 8 件事：

步骤	说明
1. 检查 Docker + Docker Compose	前置依赖校验
2. 构建 flink-ci-runner:1.20.4	CI Runner 镜像（Flink 1.20.4 + Python3 + git）
3. 启动 GitLab CE + Runner	两个容器，桥接网络
4. 预拉取 Runner Helper 镜像	防止 CI job 因网络问题拉取超时
5. 生成/上传 SSH 公钥到 GitLab	免密推送代码
6. 在 GitLab 上创建项目仓库	API 自动创建，不需要手动操作
7. 创建并注册 Project Runner	POST /api/v4/user/runners 注册
8. 配置本地 git remote gitlab	ssh://git@localhost:2224，一键推送

环境就绪后，打开 http://localhost:8929（用户名 root，密码 flink1234）就能看到 GitLab 面板。随便改个 SQL 文件，推送代码即可触发流水线：

git push gitlab main

流水线设计

架构总览

示例中搭建的 CI/CD 流水线分为了以下几个阶段：

• 规范检查： 我们的数仓有一些必要的规范，命名规范、SQL 语法规范等等，这一步是对 SQL 代码是否满足数仓规范的检查
• SQL校验： 基于 Flink 内置的 Calcite 解析、验证能力对提交的 SQL 代码进行语法检查、语义检查
  • 语法检查：检查 SQL 语法是否符合 Flink SQL 语法规范
  • 语义检查：解析 SQL 查询的 Catalog，解析表名、字段名、函数名，这些 Catalog 中都有吗？引用的对吗？类型对吗？
• 动态编排： SQL Script 的发布往往是需要编排的，比如：上游新增了一个字段需要先发 DWD 再发 ADS，搞反了会导致发布失败，因此需要有一定的编排策略（当然可以根据自己的实际情况看下是否保留这一步）
• 权限审批（未实现）： 真实的发布是需要走审批流程的，示例中为了简单没有实现这一环，读者可以根据自己的实际情况接入公司内部的审批系统
• 部署线上： 审批通过后，将 SQL Script 部署到线上（当然可能涉及到重启的方式，示例中为了简单没有实现这一环）

示例中，我们将 PR 作为触发 CI 的时机：用户提交了 PR 且涉及到了 .sql 文件的变更则触发 CI 流程。我们将合并作为触发 CD 的流程：用户合并了 PR 且合并到了 main 分支则触发 CD 流程。

当然，CD 流程只是为了演示整体的链路。实际 CD 流程可能涉及到权限、审批流、部署顺序、部署时间、部署方式等问题，读者可以根据自己的实际情况进行调整。

整条流水线由 4 个脚本和 1 个 .gitlab-ci.yml 驱动。

流水线脚本

脚本	做了什么	方式
check-warehouse-naming.sh	检查 {层级}_{业务}_{后缀}.sql 三段式命名合规性	全量扫描 warehouse/
validate-sql.sh	Flink SQL 语法 + 语义校验（表是否存在、字段是否在、类型是否匹配）	增量（git diff），CI 下自动获取变更文件
generate-deploy-pipeline.sh	检测两次 commit 间的 SQL 变更，调用 Python 编排脚本	增量
build-deploy-order.py	从文件名解析层级，按 dwd → dws → ads 排序，输出 child-pipeline.yml	被 shell 脚本调用

核心校验命令只有一条，不连 Flink 集群，2 秒出结果：

$FLINK_HOME/bin/flink run --target local $BOOTSTRAP_JAR \
  --script-file file://<sql文件> --validate

部署则是两步串联：generate-deploy-pipeline.sh 检测变更 → build-deploy-order.py 按层级排序、生成子流水线。子流水线通过 GitLab 的 trigger + artifact 机制按 stage 顺序执行，--catalog-file 注入生产环境表结构。

流水线配置

stages:
  - validate
  - rule-check
  - deploy

warehouse-naming-check:        # 全量命名规范检查
  stage: rule-check
  script: bash scripts/check-warehouse-naming.sh
  rules:
    - if: $CI_PIPELINE_SOURCE == "merge_request_event"  # MR 触发
      changes: [example-cicd/warehouse/**/*.sql]
    - if: $CI_COMMIT_BRANCH == "main"                   # main 触发
      changes: [example-cicd/warehouse/**/*.sql]

flink-sql-validate:            # 增量语法校验
  stage: validate
  script: bash scripts/validate-sql.sh
  rules:   # 同上，MR 和 main 都触发

generate-deploy-pipeline:      # 生成部署子流水线（仅 main）
  stage: deploy
  script: bash scripts/generate-deploy-pipeline.sh
  artifacts: [example-cicd/child-pipeline.yml]
  rules:
    - if: $CI_COMMIT_BRANCH == "main"
      changes: [example-cicd/warehouse/**/*.sql]

deploy-jobs:                   # 触发子流水线（仅 main）
  stage: deploy
  needs: [generate-deploy-pipeline]
  trigger:
    include:
      - artifact: example-cicd/child-pipeline.yml
        job: generate-deploy-pipeline
    strategy: depend
  rules:   # 同 generate-deploy-pipeline

几个关键设计：

• changes: 确保只有 SQL 变更才触发，改脚本、文档不跑流水线
• MR 只跑 CI（规范检查 + 语法校验），CD 仅 main 分支触发
• strategy: depend 保证子流水线挂了，父流水线也标红

本地调试

所有脚本都脱离 CI 环境变量独立可跑：

# 校验单个文件
bash scripts/validate-sql.sh --file warehouse/orders/dwd_orders_di.sql

# 模拟增量校验
CI_COMMIT_BEFORE_SHA=HEAD~2 CI_COMMIT_SHA=HEAD bash scripts/validate-sql.sh

# 手动生成部署子流水线
bash scripts/generate-deploy-pipeline.sh --from HEAD~2 --to HEAD

底层原理拆解

SQL 的多语句切分机制已在 上一篇（阶段一：智能切分）中详细讨论——六种状态的逐字符扫描、引号/注释内的分号不切分，这里不再赘述。直接聚焦切分之后发生的事情。

全链路概览

一条 flink run 命令背后，flink-sql-bootstrap 支持三种模式，依次递进：

• --validate：只走到 parse + DDL 执行 + DML 暂存，不编译不提交。耗时 ~2 秒
• --compile：比 validate 多一步 planner.compilePlan()，产出优化后的执行图 JSON
• 默认执行：完整走完 translate → injectResourceSpec → executeInternal，提交到集群

其中 --validate 是整个 CI/CD 管道中最关键的环节——它决定了 代码能不能继续往下走。下面展开它的内部链路。

--validate 内部执行链路

切分后每条 SQL 都会经历三步解析——ParserImpl.parse() 内部的三阶段流水线：

① Calcite CalciteParser → SqlNode 列表（词法 + 语法解析）

Flink 的 CalciteParser 包装了 Apache Calcite 的 SqlParser，调用 parseSqlList(statement) 将 SQL 文本解析为一个或多个 SqlNode（AST 节点）。这一步纯做语法解析，不查 Catalog。

检查内容：关键字拼写、括号匹配、SQL 语法结构是否合法。SELET 写成这样——直接抛 SqlParserException，给出行号和列号。

源码中有一个值得注意的细节：CalciteParser 会把 Calcite 内部抛出的 SqlParserEOFException（EOF 异常）包装成更友好的 SqlParserException，这就是 --validate 报错时看到的那些精确错误信息。

② SqlNodeToOperationConversion → Operation（语义校验 + 转换）

这是整个 validate 链路最重的环节。Flink 通过 SqlNodeToOperationConversion.convert() 完成了两件事：

语义校验：先调 FlinkPlannerImpl.validate(sqlNode)，这是 Calcite 内置的语义验证器，负责：

• 标识符解析——SELECT * FROM orders 中的 orders 是不是已经在 Catalog 里了？
• 列引用校验——* 展开后，每个列都在表结构中吗？
• 类型检查——SUM(amount) 中 amount 是可聚合的类型吗？
• 子查询合法性、JOIN 条件合理性等

类型转换：验证通过后，将合法的 SqlNode 转换为 Flink 的 Operation 对象。这一步使用了注册式转换器（SqlNodeConverters）——不是写死的 instanceof 判断，而是在初始化时注册所有已知的转换器（SqlCreateTableConverter、SqlQueryConverter 等），按 SqlNode 的实际类型自动匹配并调用对应的转换器。当需要新增 SQL 语句类型时，只需注册一个新转换器，整个 parse 链路零改动。

③ 按 Operation 类型分派

parse 完成后，根据返回的 Operation 类型决定执行策略：

Operation 类型	具体子类	策略	原因
DDL 建表/视图/函数	CreateTableOperation / CreateViewOperation / CreateFunctionOperation	立即执行	后续语句 parse 时依赖内存 Catalog
环境配置	SetOperation / ResetOperation	立即执行	影响编译环境
DML 写操作	SinkModifyOperation (INSERT/UPDATE/DELETE)	暂存，不编译	validate 不需要出执行计划
批量 DML	StatementSetOperation	暂存，不编译	同上，内部包裹多个 ModifyOperation

为什么 DDL 必须立即执行？ 这是一个关键设计。看这个实际例子：

CREATE TEMPORARY TABLE ods_orders (order_id STRING, amount DECIMAL(10,2), ...);
CREATE TEMPORARY TABLE dwd_orders (...);
INSERT INTO dwd_orders SELECT ... FROM ods_orders ...;

如果 CREATE TABLE ods_orders 像 DML 一样暂存不执行，当 parser 解析第三条 INSERT ... FROM ods_orders 时，Catalog 里根本没有 ods_orders 这张表——第二步的语义检查直接报 Table 'ods_orders' not found。

DDL 的副作用（修改 Catalog）必须对所有后续语句立即可见。这和编程语言中「先 import 才能使用」是一样的道理。

三种模式对比

	--validate	--compile	默认执行
切分 + parse + DDL 执行	✅	✅	✅
DML 暂存	✅	✅	✅
compilePlan() 编译	❌	✅	✅
translatePlan() → 提交	❌	❌	✅
耗时	~2s	~5s	取决于数据量
需要 Flink 集群？	否	否	是

小结

本系列写了两篇文章，上一篇是 Flink 实时数仓开发实战：像 Hive 那样用 Flink SQL。这个系列的初衷是希望结合我自己之前的经历和经验，给大家提供一种有效、可靠、低成本的 Flink SQL 数仓研发流程，从而提升研发效率和质量。

作者本人非常热爱 Coding，热爱数据研发、热爱生活，希望在和大家讨论中共同进步，我乐意和大家一起交流，回答大家所提的 任何问题。也可以连线讨论任何我能力圈范围内的问题（包括但不限于：研发效率、AI、数据开发、历史文学，可能还有价值投资）。

本文基于 Flink SQL Bootstrap v1.0.0 及 example-cicd 实战示例

关于作者

🙋 前阿里巴巴数据研发工程师，专注实时引擎、实时平台、实时应用开发。

👏 欢迎反馈和交流实时应用开发中的任何问题，我将尽我所能帮助大家。如何联系我：

• 项目中提交 Issue
• Email：tonyabasy@163.com

👏 同时欢迎大家参与 flink-sql-bootstrap 共建。