数仓AI提效之Skill篇

当我把本地 AI 接进数据中台后

最近我把本地 AI 接到了数据中台上。
我把本地 AI 接入数据中台后，用一组 Skill 串起了拉代码、查表、看血缘、生成校验、跑 Hive、读日志的流程。
AI 负责搬上下文和整理结果，我负责确认口径和风险。

如果你也是做数仓、数据开发、BI 支撑，或者经常和 SQL、调度任务、Hive/Spark 日志打交道，应该很熟悉这种状态：

打开数据开发平台 找到对应 Job 看生产版本 看待发布版本 复制脚本到本地 查目标表结构 查字段备注 查上游表有没有这个字段 看任务依赖 改 SQL 生成校验 SQL 提交测试 等执行结果 失败了再去翻日志 从日志里找真正有用的几行 回来继续改

太碎了，

碎到一个需求的业务逻辑可能十分钟就想明白了，最后却要在本地编辑器、数据中台、查询平台、跳板机、日志页面之间切半天。

我这次想解决的，就是这个问题。

AI 负责查上下文、整理资料、跑校验、读日志。
我负责确认口径、判断风险、决定是否继续。

一、我接的不是一个工具，而是一组 Skill

一开始我也只是想让 AI 帮我跑一下本地 SQL。

后来发现，只能跑 SQL 还不够。

真实的数据开发不是“写一段 SQL 然后执行”这么简单。它前面有作业查找，中间有版本差异，后面有校验和日志分析。

所以我最后没有做成一个单点工具，而是拆成了一组 Skill。

1. 拉取任务代码

有时候我只知道一个 Job 名，需要先把这个 Job 下的任务脚本拉到本地。

这个流程以前要自己去数据中台找脚本，然后在本地建文件夹，一个 Task 一个 Task 往下搬。尤其是 Task 多的时候，特别痛苦！

现在 Skill 会按固定步骤处理：

确认 job_name 检查本地目录是否已存在 检查远端目录是否已存在 如有冲突先问我 在跳板机执行受控脚本 生成任务脚本和血缘信息 拉回本地 输出本地目录和远端目录

这里我刻意加了冲突确认。

因为拉任务代码看起来是只读，但如果本地已有同名目录，直接覆盖就可能把我正在改的东西冲掉。

所以本地或远端目录已存在时，必须先停下来问我。

2. 表级开发助手

很多需求不是从 Job 开始的，而是从一张表开始的。

比如：

这张表在哪里产出？ 谁在用这张表？ 这个字段从哪里来？ 生产版本和待发布版本差了什么？ 这次改口径会影响哪些下游？

这类问题如果完全靠人工查，最容易漏。

所以我把它拆成表级开发卡片的流程。

AI 会优先从本地的yh-data-develop目录里找资料，包括：

• Job / Task 脚本；
• 生产版本和待发布版本；
• 参数配置和任务依赖；
• 表结构和 insert 逻辑；
• 下游消费位置。

然后整理成一张卡片：

表名是什么 它是产出表、来源表还是中间表 由哪个 Job / Task 产出 生产脚本在哪里 待发布脚本在哪里 有哪些下游消费 参数和依赖是什么 字段口径是什么 建议怎么测试 风险点有哪些

这个 Skill 对我来说很有用。

因为它不是直接上来就改 SQL，而是先把“这张表现在是什么情况”讲清楚。

很多时候，真正减少风险的不是改得快，而是改之前先知道自己在动哪条链路。

3. 迭代开发流程

如果要新增字段、调整口径、改链路，就不能直接动文件。

我给这类需求单独定了一套迭代开发规范：

先读目标表结构 再读产出 SQL 再看生产版本和待发布版本 再看上游依赖 再看下游使用 缺资料先补资料 然后输出迭代方案 等我确认 确认后才改待发布版本 改完再生成校验 SQL 必要时跑 dev 表验证

这里有几个硬规则：

• 涉及 Pallas / YH Data Studio 的离线任务时，以本地yh-data-develop目录为准；
• 默认只改待发布版本，不直接改生产版本；
• 新增字段只能追加在字段列表最后，不能为了“看起来更顺”去调整已有字段顺序。

字段顺序这个规则很重要。

在数仓里，字段顺序不是排版问题。很多时候调整字段顺序意味着删表、重建表，或者影响 insert select 的位置映射，风险很高。

所以我宁愿让新增字段放在最后，也不允许 AI 自作聪明地重排字段。

4. 数据校验 SQL 生成

改完 SQL 以后，还要校验。

以前这些校验经常要手写：

行数对比 主键重复 空值率 金额汇总 数量汇总 dev/pro 对比 按维度对比

现在我把校验生成也做成了规范。

它不会拿到一个表名就乱猜字段，而是先找 DDL 或建表 SQL，从里面识别数值型字段，再判断哪些字段适合作为指标。

金额、数量、客流、毛利、优惠、积分这类字段可以作为校验指标。

但像shop_id、goods_id、stat_flag这种虽然也是数字，不能直接当指标汇总。它们更适合作为维度或过滤条件，用来做分组对比。

所以生成前会先给我一个确认：

识别到的校验指标有哪些 哪些字段适合作为维度 排除掉哪些数值字段 排除原因是什么 校验日期是什么 按哪些维度校验

我确认后，再生成 base 校验、维度校验或者 dev/pro 对比校验。

比如 base 校验看整体行数和指标汇总；维度校验可以按sdt、shop_id、goods_id、bd_id、catg_l_id这类字段分组，看不同维度下 dev/pro 或新旧逻辑的差异。

这一步看起来只是生成 SQL，但实际省了不少时间。因为数据开发里很多校验不是难写，而是每次都要重复写一遍。

5. 本地 SQL 自动跑 Hive，并拉回日志

这是我最早做、也是用得最多的一段。

本地写完 SQL 后，我可以让 AI 上传到 JupterNote，通过固定的 PySpark runner 跑 Hive。

中间链路大概是这样：

本地 IDE / 本地 AI ↓ Skill：检查 SQL、识别风险、上传文件、拉回结果 ↓ JupterNote / 跳板机 ↓ PySpark runner ↓ Hive / Spark / 数据中台元数据

对应的流程是：

检查本地 SQL 文件 检查远端 runner 是否存在 识别 SQL 是否有写入风险 如果是写入，先说明目标表并问我 上传 SQL 到远端运行目录 执行 Kerberos 认证 调用固定 PySpark runner 拉回 result.json 拉回 stdout.log / stderr.log 分析执行结果 失败时定位原因 需要重跑前再次问我

我不希望 AI 自己拼一堆不可控命令，所以远端只允许调用固定 runner。

这个 runner 做的事情很简单：接收一个 SQL 文件，按顺序执行每条 SQL，并把状态写到result.json。

执行开始时先写：

{"status":"running","allow_writes":false,"statements":[]}

每条 SQL 执行时再记录：

{"index":1,"status":"success","columns":["xxx","yyy"],"row_count":10,"duration_seconds":3.21}

只读 SQL 默认只取前 100 行。

这样我能看到字段和值，但不会把大结果集拉回本地。

如果失败，AI 会优先读结构化的result.json，再去看 stdout 和 stderr。

这比直接把几千行日志丢给 AI 稳定很多。

最后AI会给我这样的结论：

第 2 条 SQL 执行失败。 核心原因是字段 xxx 在当前子查询中没有产出。 最终 select 引用了这个字段，但上游临时结果没有保留。 建议先确认字段来源；如果字段确实来自明细层，需要在中间聚合层补出来。

二、完整流程跑起来是什么样

现在这套东西串起来以后，一个表迭代需求大概会变成这样。

比如我要给某张销售主题表新增一个指标。

以前我可能会这样做：

自己找任务 自己复制脚本 自己查字段 自己比生产和待发布 自己改 SQL 自己提交测试 自己翻日志 自己写校验 自己再改

现在我更愿意这样说：

帮我看一下这张表的待发布版本，新增这个指标。 先查字段来源和上下游影响。 按项目 SQL 规范出一个迭代方案。 确认后再改待发布版本。 改完生成校验 SQL，先跑 dev 表验证。 如果失败，把日志拉回来分析，重跑前先问我。

这句话背后，其实已经包含了一整条流程：

1. AI 先找资料，而不是直接改；
2. 资料不够时，先补表结构、查元数据、看任务依赖；
3. 确认方案后，才动待发布版本；
4. 改完生成校验 SQL，再把本地 SQL 上传到远端执行；
5. 执行失败时，拉回result.json、stdout.log、stderr.log，先判断失败在哪条 SQL，再结合日志分析原因；
6. 如果要重跑，必须再问我。

这个过程里，我不用一直搬上下文。

我只需要在几个关键点做判断：

字段口径是否正确 来源表是否可靠 下游影响能否接受 失败后的修复方向是否合理 是否继续重跑

对我来说，这才是人应该花精力的地方。

三、最关键的不是模型，而是流程结构化

这套东西做下来，我越来越觉得，AI 在数据开发里好不好用，不只取决于模型会不会写 SQL。

更关键的是流程有没有结构化。

我现在会尽量把流程拆成几个固定边界。

1. 固定作业来源

涉及 Pallas / YH Data Studio 的离线任务时，以yh-data-develop为权威目录。

如果同一个脚本在别的目录也有，优先看这里。

这样 AI 不会拿旧目录里的脚本当最新版本来改。

2. 固定输出目录

AI 自动生成的中间资料，不要散落在业务目录里。

比如：

z_ai_meta/show_create_sql/ 存 show create table 临时 SQL z_ai_meta/hive_runs/ 存每次 Hive 执行结果和日志 z_ai_meta/table_structures/ 存临时拉回的表结构

业务目录只放最终确认要归档的东西。

这样后面清理和追溯都容易很多。

3. 固定结果格式

远端执行结果必须有结构化文件，也就是result.json。

日志可以保留，但不能只依赖日志。

因为日志太长、太杂，而且很多错误埋在中间。有了结构化结果以后，AI 可以先知道：

总体成功还是失败 失败在第几条 SQL 每条 SQL 的预览是什么 只读查询返回了哪些字段 样例数据是什么 初步错误是什么

然后再去日志里补证据。

这个设计比“执行失败，请看日志”有用太多。

四、安全边界：我是怎么界定的

AI 一旦能提交 SQL，安全边界就必须先画清楚。

我这里的界定方式很简单：先按操作分级，再按目标库限制，最后把是否执行交还给人确认。

更关键的是，这个限制不是只写在提示词里，而是写进了 runner 代码里。

也就是说，就算我在对话里允许执行，如果 SQL 目标是正式库，而 runner 代码没有改，它也动不了正式库。

1. 先把 SQL 分成三类

第一类是只读 SQL。

比如select、show create table这类，只读取元数据或样例结果，不改数据。

这类可以让 AI 自动跑，但结果必须受控：只读查询默认只取前 100 行，避免把大结果集拉回本地。

第二类是开发库写入。

比如写dev、tmp、analyse_tmp这类开发库或临时库。

这类不是完全禁止，但必须先识别目标表、说明风险、让我确认。确认后才会给 runner 传--allow-writes。

第三类是高风险操作。

比如写生产库、删表、结构变更，或者无法识别目标表。

这类默认直接停止，不让 AI 自己继续。

2. 代码里先做第一层拦截

我在 runner 里先写了一层基础规则：

ALLOWED_WRITE_DBS=("dev","tmp","analyse_tmp")WRITE_RE=re.compile(r"^\s*(insert|create|truncate|drop)\b",re.IGNORECASE,)BLOCKED_RE=re.compile(r"^\s*alter\b",re.IGNORECASE,)

判断逻辑也很直接：

defvalidate_statement(statement,allow_writes):ifBLOCKED_RE.search(statement):raiseValueError("alter is not allowed by default")ifnotWRITE_RE.search(statement):returnifnotallow_writes:raiseValueError("write statement found, but allow_writes is false")match=TARGET_TABLE_RE.search(statement)ifnotmatch:raiseValueError("write target table could not be detected")table_name=match.group(1).replace("`","").strip()if"."notintable_name:raiseValueError("write target must include database name")db_name=table_name.split(".",1)[0].lower()ifdb_namenotinALLOWED_WRITE_DBS:raiseValueError("write target database is not allowed")

这里有几个关键点：

• SQL 里出现insert、create、truncate、drop，先按写入/删除处理；
• 写入目标必须带库名，不能只写表名；
• 目标库只能是dev、tmp、analyse_tmp；
• alter默认禁止；
• 即使目标库合法，如果没有人工确认，也不会执行；
• 如果目标库是正式库，即使人工确认了也会被 runner 拦住，除非先改代码里的白名单。

3. 人工确认是最后一道门

这层 Regex 校验不是为了证明“绝对安全”。

它只是第一层防线：先把明显的写入、删除、结构变更拦住，让 AI 不会默认执行高风险 SQL。

我真正想要的是这条边界：

读操作：AI 可以自动做，但结果要限量。 开发库写入：AI 可以准备，但执行前必须问我。 生产库写入：默认不允许。 结构变更：默认不允许。 目标识别不清楚：默认不执行。

也就是说，AI 可以把风险识别出来、把目标表告诉我、把命令准备好。

但是否执行，由我决定。

而且这个“由我决定”也不是无限放权。我的确认只能放行dev、tmp、analyse_tmp这类代码白名单里的库；如果 SQL 指向正式库，runner 还是会直接拒绝。

后面如果继续加强，可以再加 SQL Parser / AST 分析、权限控制、库表白名单、操作审计。

但目前这层边界已经够挡住大部分误操作。

五、项目规范必须写进去

还有一个很现实的点：项目规范一定要写进去。

数据开发里很多规范，AI 不会天然知道。

比如我们写 SQL 会要求：

SQL 关键字小写 表名必须带库名 逗号放在字段前面 中文别名要加单引号 as 要对齐 select 后换行 分区表一定要加分区 新增字段只能追加到最后 默认只改待发布版本 Spark insert 先写 dev 表验证

这些看起来像格式问题，但在真实数仓里经常和风险有关。

如果 AI 为了“语义更顺”调整字段位置，可能会引出很麻烦的问题。

所以我会把规则写死。

每次修改前都按规则来，比每次口头提醒稳定得多。

六、现在已经覆盖了哪些场景

这套流程现在已经不是一个“自动跑 Hive”的小工具了。

它基本覆盖了我日常数据开发里最容易被打断的几类动作：

按 Job 拉取任务脚本 整理表级开发资料 查生产版本和待发布版本差异 查表结构和字段来源 分析上下游影响 输出迭代方案 按确认后的方案修改待发布版本 生成 base / 维度 / dev-pro 校验 SQL 上传本地 SQL 到 JupterNote 跑 Hive 拉回 result.json 和日志 分析失败原因 确认后重跑

七、它真正省下来的是什么

它省下来的不只是几行命令。

更重要的是，少了很多一层一层、找血缘、读代码的时间。

以前排查一个问题，经常要顺着链路往上找：

这个字段从哪来？ 上游哪张表产出？ 上游 Task 在哪里？ 生产和待发布有没有差异？ 下游还有谁在用？ 日志里的报错到底对应哪段 SQL？

这些事人当然也能做，但很费时间，而且特别容易被打断思路。

现在可以让 AI 先去找。

它读代码、扫目录、查依赖的速度比我快很多。尤其是任务多、链路长的时候，它可以先把相关脚本、字段来源、上下游影响和日志重点整理出来。

我不用从头翻到尾，只需要看它整理后的结论，再判断：

字段来源对不对 口径有没有歧义 下游影响能不能接受 这次失败是代码问题、数据问题，还是资源问题 是否需要继续重跑

所以我的工作重心从“找作业、搬上下文”，变成了“判断结果、确认风险”。

八、最后

AI真的太强了，我是AI孝子。