当前位置：首页 > news >正文

用 AI 改造一个 Flink SQL 项目：从脚本提交到数据同步平台

news 2026/6/18 11:04:05

用 AI 改造一个 Flink SQL 项目：从脚本提交到数据同步平台

这几年做实时数仓和 Flink SQL 任务时，我越来越明显地感觉到一件事：

很多数据同步任务本身并不复杂，复杂的是配置、提交、排查和维护这一整套流程。

比如一个很普通的 MySQL 到 Kafka 同步任务，真正的业务逻辑可能只有几行字段映射。但在落地的时候，我们还是要配置数据源、确认表结构、写 source DDL、写 sink DDL、写 insert SQL、配置并行度、配置 checkpoint、提交到 Yarn、查看 Flink 任务状态、排查失败原因。

这些事情单独看都不难，但组合起来就很容易变成重复劳动。

所以最近我准备把手上的一个 Flink SQL 项目重新改造一下，借助 AI 搓一个轻量级的数据同步平台。

这个项目叫 sqlSubmit。

它最早的定位很简单：把写好的 Flink SQL 提交到集群里跑起来。

目前项目里已经有一些基础能力：

支持读取 SQL 文件并提交执行。
支持通过 properties 管理任务参数。
支持 checkpoint、state backend 等运行配置。
支持注册 UDF、UDAF、UDTF。
已经扩展了一些自定义 connector，比如 MySQL、HBase、Redis、StarRocks、HTTP、Socket。
项目里也沉淀了不少 Flink SQL 示例，包括 Kafka、JDBC、Hudi、Iceberg、窗口计算、lookup join 等场景。

也就是说，它不是一个空项目，而是一个已经能跑 SQL 任务的工具型项目。

这次改造的目标，就是把它从“脚本提交工具”逐步演进成“可配置、可生成、可提交、可追踪”的同步平台。

01 为什么要做平台化

用脚本提交 Flink SQL 有一个好处：简单直接。

开发同学写好 SQL，本地或者服务器上执行提交脚本，任务就能跑起来。对于个人开发、功能验证、小规模任务来说，这种方式很顺手。

但随着任务数量变多，问题也会慢慢出现：

每个任务都要手写 source 和 sink DDL。
字段类型映射容易出错。
不同任务的 checkpoint、并行度、队列等参数缺少统一管理。
任务提交之后，很难从平台视角追踪运行状态。
SQL 改过几版、线上跑的是哪一版，不容易回溯。
新同学接手时，需要先理解一堆脚本、参数和目录约定。

这些问题的本质，不是 Flink SQL 不够好，而是缺少一层工程化的平台能力。

所以我希望做一个平台，把重复的部分收敛起来：

页面上配置数据源，平台采集元数据，用户选择字段映射，系统生成 Flink SQL，然后复用现有 sqlSubmit.jar 提交到 Yarn，最后通过 Flink REST API 回查任务状态。

这样既保留 Flink SQL 的透明度，又降低日常同步任务的使用门槛。

02 第一阶段先做什么

平台化最怕一开始就做大而全。

如果一上来就把 CDC、血缘、多租户、权限、拖拽编排、指标大盘全部拉进来，最后很容易变成一个看起来很完整、但主链路还没跑稳的系统。

所以第一阶段我只打算做一个 MVP。

先支持几条最基础的同步链路：

datagen -> print
datagen -> kafka
mysql -> print
mysql -> kafka
mysql -> mysql

Kafka 第一阶段只作为目标端，不先做 source schema 推断。

MySQL CDC、Kafka source、权限、多租户、数据血缘、可视化拖拽编排，这些能力都先放到后面。

第一阶段真正要验证的是：

一个同步任务，能不能从页面配置开始，经过 SQL 生成、版本保存、任务提交、状态回查，完整跑通。

只要这条主链路跑通，后面的能力就可以一层一层加。

03 整体架构

第一版架构会保持轻量。

前端用 Vue，后端用 Spring Boot，元数据库用 MySQL，执行层继续复用当前项目的 sqlSubmit.jar。

整体流程大概是这样：

用户在页面创建数据源、选择来源表、配置目标端和字段映射。

后端保存结构化任务配置，并根据配置生成 Flink SQL。

生成后的 SQL 会保存版本，每次提交都指向一个不可变的 SQL 版本。

提交时，平台把 SQL 文件和任务 properties 文件写到指定目录，然后拼接 Flink CLI 命令，把任务提交到 Yarn。

任务运行后，平台通过 Flink REST API 查询任务状态、异常信息和 checkpoint 概况。

这里有一个关键取舍：

第一阶段不重写执行引擎。

因为当前 sqlSubmit 已经具备 SQL 文件解析、参数加载、checkpoint 配置、UDF 注册、StatementSet 执行等能力。平台侧真正要做的，是把“任务配置、SQL 生成、提交编排、状态追踪”这几件事补齐。

这样改造成本更低，风险也更可控。

04 平台核心模块

第一版我会把后端拆成几个模块。

第一个是数据源管理。

平台需要维护 MySQL、Kafka、Datagen、Print 这些数据源。MySQL 要支持 JDBC 连接测试，Kafka 要验证 bootstrap server 和 topic，Datagen 和 Print 作为内置数据源直接可用。

第二个是元数据采集。

对于 MySQL source，平台可以通过 JDBC 元数据或者 SHOW FULL COLUMNS 读取表字段，然后把 MySQL 类型映射成 Flink SQL 类型。

比如：

bigint 映射为 BIGINT
int 映射为 INT
varchar、text 映射为 STRING
decimal(p,s) 映射为 DECIMAL(p,s)
datetime、timestamp 映射为 TIMESTAMP(3)

如果遇到不能安全映射的类型，先映射成 STRING，同时标记为需要人工确认。

第三个是 SQL 生成器。

这是整个平台的核心。

用户在页面上选择 source、sink、字段映射和运行参数后，平台生成完整的 Flink SQL：

CREATE TABLE source_xxx (...);CREATE TABLE sink_xxx (...);INSERT INTO sink_xxx
SELECT ...
FROM source_xxx;

生成 SQL 不是为了隐藏 SQL，而是为了让 SQL 更稳定、更可审查。

每次提交前都能预览，每次提交后都保存一个不可变版本。这样后续排查问题时，可以明确知道某一次任务到底跑的是哪一版 SQL。

第四个是任务管理。

任务需要区分草稿、已生成、运行中、失败、取消、完成等状态。

每次生成 SQL，都可以保存为一个版本；每次提交，都会产生一个任务实例。

任务定义、SQL 版本、运行实例要拆开存。

这样做的好处是，任务可以持续编辑，但已经提交过的 SQL 版本不能被偷偷改掉。

第五个是 Yarn 提交器。

平台后端会生成 SQL 文件和 properties 文件，然后复用现有 sqlSubmit.jar 提交到 Yarn。

示例命令大概是这样：

flink run \-m yarn-cluster \-ynm user_job_name \-yqu default \/opt/sqlsubmit/sqlSubmit.jar \--sql /opt/sqlsubmit/generated/sql/job_1001_v1.sql \--job.prop.file /opt/sqlsubmit/generated/prop/job_1001_v1.properties

第六个是 Flink 状态回查。

任务提交之后，平台需要记录 Yarn application id、Flink job id、提交命令、SQL 路径、任务状态、异常摘要等信息。

后续通过 Flink REST API 查询运行状态、异常信息和 checkpoint 概况。