---
name: pyspark-sql-guardrails
description: 在编写或评审 PySpark 脚本、Spark SQL 字符串、`spark.sql` 调用、从配置文件加载的 SQL、ETL 作业、数仓维护脚本,或任何必须把 SQL 限制为 SELECT 和 INSERT 的数据流水线代码时使用。
---

# PySpark SQL Guardrails(PySpark SQL 防护栏)

## 校验关口(先读这一段 — 不可妥协)

**本 skill 产出的每一条 SQL 字符串,在跑过自带的校验器之前都是不可信的。** 校验器只需要一条命令,在 SQL 生成之后、**展示给用户之前**立即执行。不存在绕过此步骤的路径。

整条流程就是两条命令 + 一个条件:

```bash
# 第 1 步 — 生成 SQL(脑中起草 / 写出来)
# 第 2 步 — 立即把 SQL 喂给校验器(在同一轮里跑)
echo "<the SQL you just wrote>" | python3 skills/pyspark-sql-guardrails/scripts/validate_sql.py

# 第 3 步 — 读输出:
#   PASS - first keyword: select, body length: N   → 可以把 SQL 给用户看了
#   FAIL - <reason>                                → 停下,修 SQL,再跑一次
```

**如果你还没把 `PASS -` 这一行贴到对话里,就不允许贴 SQL。** 校验器就是关口,SQL 站在关口的另一边,绝不能空着手闯关。

这条规则由本 skill 自己强制,而非外部系统。遵守它是任何含 SQL 字符串的输出的前提。

---

## 概述(Overview)

本环境下的 PySpark SQL 只接受白名单:生成或执行的 SQL 只能是 `SELECT` 或 `INSERT`。在通过校验之前,所有 SQL 字符串、格式化后的 SQL 模板、从配置加载的 SQL、notebook 单元格,以及 `spark.sql(...)` 调用一律视为不安全。

核心原则:**绝不能靠"看上去不像危险"来执行 SQL;只有在证明它就是一条以 SELECT 或 INSERT 开头的、合法的语句后,才能执行。**

## 强制校验(Hard Rule)

**每条 SQL 字符串在到达 `spark.sql(...)` 之前都必须通过 `assert_select_or_insert()` — 不许有例外。** 这条规则不可妥协,适用于:

- 你刚生成的 SQL(字面量、f-string、`.format(...)`、`+` 拼接)
- 从 YAML / JSON / INI / 环境变量 / 命令行 / 配置文件加载的 SQL
- 通过参数、函数参数、notebook 变量传入的 SQL
- 从聊天消息里粘贴进来的 SQL
- 由模板引擎(Jinja、string.Template 等)产出的 SQL

唯一可接受的调用形式:

```python
# 推荐:封装函数,校验+执行一步到位
safe_spark_sql(spark, sql)

# 或者:先校验,再显式执行
assert_select_or_insert(sql)        # 违规即抛错
spark.sql(sql)
```

**禁止的调用形式(出现即视为流水线失败):**

```python
spark.sql(sql)                                # 直接调用,未校验
spark.sql(f"SELECT ... {user_input} ...")     # f-string 直接进 spark.sql
spark.sql(config["sql"])                      # 配置驱动的 SQL 未校验
spark.sql(open("queries/xxx.sql").read())     # 从文件加载的 SQL 未校验
```

如果 `assert_select_or_insert()` 抛错,流水线立即停止。**不要**削弱规则,**不要**用正则去剥除禁用关键字,**不要**把 SQL"改写"成看似安全的样子。要么拒绝,要么重新生成。

## 必守规则(Required Rule)

**允许:**
- `SELECT ...`
- `WITH ... SELECT ...`
- `INSERT INTO ... SELECT ...`
- `INSERT OVERWRITE ... SELECT ...` — 仅当用户明确允许本次任务使用 overwrite,否则必须先问

**禁止**(即便是维护或元数据刷新):
- `DELETE`、`TRUNCATE`、`DROP`、`ALTER`、`CREATE`、`REPLACE`、`MERGE`、`UPDATE`
- `MSCK REPAIR`、`REFRESH`、`ANALYZE`、`CACHE`、`UNCACHE`、`VACUUM`、`OPTIMIZE`
- 用分号分隔的多条语句
- 从 YAML/JSON/env/CLI 来的、未经任何处理直接喂给 `spark.sql` 的 SQL

**执行关口:** 每次 `spark.sql` 调用都必须先调 `assert_select_or_insert(sql)`(或 `safe_spark_sql(spark, sql)`)。没有快速通道。详见上面的 *强制校验(Hard Rule)*。

## 速查表(Quick Reference)

| 场景 | 做法 |
|---|---|
| 想要取数 | `safe_spark_sql(spark, "SELECT ...")` |
| 想要写数 | `safe_spark_sql(spark, "INSERT INTO table SELECT ...")` |
| 想要清理/去重/删除 | 用 SELECT 派生干净数据,再用 `safe_spark_sql` + INSERT 写入已批准的目标/staging 表;**不要**直接删除旧数据 |
| 想要做 schema/表/分区维护 | 停下来询问用户;**不要**输出 DDL 或元数据 SQL |
| SQL 来自配置文件 | 先 `assert_select_or_insert(sql)`,**再** `spark.sql(sql)` |
| 用户要求"快速修一下" | 安全规则照旧 — 先跑 `assert_select_or_insert` |
| SQL 由 `sql-context-builder` 生成 | 执行前必须跑 `assert_select_or_insert` |

## 安全模板(Safe Pattern)

**必须:** 每个 `spark.sql` 调用点都走 `safe_spark_sql`(或在 `spark.sql` 之前立即调 `assert_select_or_insert`)。在校验器面前,一条生成的 SQL 和集群之间只隔这一道闸。

在 SQL 进入执行的那条边界做校验:

```python
import re

_FORBIDDEN_SQL = re.compile(
    r"\b(delete|truncate|drop|alter|create|replace|merge|update|msck|refresh|analyze|cache|uncache|vacuum|optimize)\b",
    re.IGNORECASE,
)

_LEADING_COMMENTS = re.compile(r"\A\s*(?:--[^\n]*\n|/\*.*?\*/\s*)*", re.DOTALL)


def assert_select_or_insert(sql: str) -> str:
    text = sql.strip()
    if not text:
        raise ValueError("SQL is empty")

    # 允许一个可选的尾随分号,但拒绝多条堆叠的语句
    body = text[:-1].strip() if text.endswith(";") else text
    if ";" in body:
        raise ValueError("Multiple SQL statements are not allowed")

    normalized = _LEADING_COMMENTS.sub("", body).lstrip()
    first = normalized.split(None, 1)[0].lower() if normalized else ""

    if first not in {"select", "with", "insert"}:
        raise ValueError(f"Only SELECT and INSERT SQL are allowed, got {first!r}")

    if _FORBIDDEN_SQL.search(normalized):
        raise ValueError("Forbidden SQL keyword found")

    if first == "with" and not re.search(r"\bselect\b", normalized, re.IGNORECASE):
        raise ValueError("WITH statements must be SELECT queries")

    return body


def safe_spark_sql(spark, sql: str):
    return spark.sql(assert_select_or_insert(sql))

# 正确
safe_spark_sql(spark, """
INSERT INTO analytics.daily_customer_snapshot
SELECT * FROM staging.daily_customer_snapshot
""")

# 错误:在 Spark 看到 SQL 之前就抛错
safe_spark_sql(spark, "ALTER TABLE analytics.daily_customer_snapshot DROP PARTITION (dt='2026-06-10')")
```

如果项目里已经有 `sqlglot` 之类的 SQL 解析器,优先用 AST 校验而非正则。但白名单要保持一致:只能有一条语句、顶层是 SELECT 或 INSERT、不能出现任何禁用的 DDL/DML/维护命令。

## 本地验证脚手架(每次生成 SQL 后都要跑)

校验器如果从来不被运行,就是废物。每当产出一条 SQL 字符串(无论由你、子 agent、模板或工具产出),下一步就是在 Python 里用 `assert_select_or_insert` 跑一遍,然后才把结果给用户看。

本 skill 把自己的校验器和测试集内置在 skill 目录里,使契约自包含且与 skill 一起版本化。**不要**在调用点重新定义正则,永远从内置文件 import。

```
skills/pyspark-sql-guardrails/
├── SKILL.md           ← 本文件
└── scripts/
    ├── sql_guard.py       ← 权威校验器(assert_select_or_insert + safe_spark_sql)
    ├── validate_sql.py    ← 一行 CLI:管道喂 SQL,输出 PASS/FAIL
    └── test_sql_guard.py  ← 18 用例的标准测试集
```

### 第 1 步 — 用内置校验器(不复制、不重写)

skill 自带一个一行 CLI:`validate_sql.py`。用它。不要写不同的调用方式,不要贴内联正则,不要让子 agent 直接调 `assert_select_or_insert`,除非这个 CLI 也坏了。CLI 是 `assert_select_or_insert` 的薄壳,它调用的函数才是真正的权威。

**两种把 SQL 喂给校验器的方法:**

```bash
# 方法 A(推荐,适合多行):走 stdin
cat <<'EOF' | python3 skills/pyspark-sql-guardrails/scripts/validate_sql.py
SELECT
    c.city AS city,
    SUM(o.loan_amount) AS total_loan
FROM loan_order o
INNER JOIN customer_info c ON o.customer_id = c.customer_id
WHERE o.status = 'SUCCESS'
GROUP BY c.city
EOF

# 方法 B(仅适合单行):作为第一个参数传入
python3 skills/pyspark-sql-guardrails/scripts/validate_sql.py "SELECT 1 FROM dual"
```

**退出码:**
- `0` → `PASS - first keyword: <verb>, body length: <N>`
- `1` → `FAIL - <reason>`(reason 会指明被违反的规则)
- `2` → 用法错误(没有提供 SQL)

内置的 `sql_guard.py` 和 `validate_sql.py` 一并展示:

```python
# sql_guard.py  (内置)
import re

_FORBIDDEN_SQL = re.compile(
    r"\b(delete|truncate|drop|alter|create|replace|merge|update|msck|refresh|analyze|cache|uncache|vacuum|optimize)\b",
    re.IGNORECASE,
)
_LEADING_COMMENTS = re.compile(r"\A\s*(?:--[^\n]*\n|/\*.*?\*/\s*)*", re.DOTALL)


def assert_select_or_insert(sql: str) -> str:
    text = sql.strip()
    if not text:
        raise ValueError("SQL is empty")
    body = text[:-1].strip() if text.endswith(";") else text
    if ";" in body:
        raise ValueError("Multiple SQL statements are not allowed")
    normalized = _LEADING_COMMENTS.sub("", body).lstrip()
    first = normalized.split(None, 1)[0].lower() if normalized else ""
    if first not in {"select", "with", "insert"}:
        raise ValueError(f"Only SELECT and INSERT SQL are allowed, got {first!r}")
    if _FORBIDDEN_SQL.search(normalized):
        raise ValueError("Forbidden SQL keyword found")
    if first == "with" and not re.search(r"\bselect\b", normalized, re.IGNORECASE):
        raise ValueError("WITH statements must be SELECT queries")
    return body


def safe_spark_sql(spark, sql: str):
    return spark.sql(assert_select_or_insert(sql))


# validate_sql.py  (内置,精简版)
import sys, os
SKILL_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, SKILL_DIR)
from sql_guard import assert_select_or_insert

def main():
    sql = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
    try:
        body = assert_select_or_insert(sql)
    except ValueError as e:
        print(f"FAIL - {e}"); return 1
    first = body.split(None, 1)[0].lower()
    print(f"PASS - first keyword: {first}, body length: {len(body)}")
    return 0
```

### 第 2 步 — 每次生成 SQL 后的必用调用模式

同一条命令,每次都用,SQL 改成你刚生成的那一条。不许变体,不许走捷径,不许说"待会儿再跑":

```bash
echo "<THE_SQL_YOU_JUST_GENERATED>" | python3 skills/pyspark-sql-guardrails/scripts/validate_sql.py
```

**先把校验器的输出贴给用户,再贴 SQL。** 预期的回复样式:

```text
PASS - first keyword: select, body length: 420
```

如果出现的是 `FAIL - <reason>`,SQL 不允许通过关口。把失败原样报给用户,修 SQL,再跑校验器。循环到 `PASS` 为止。

### 第 3 步 — 标准测试集(每次改校验器时跑)

skill 自带 `test_sql_guard.py`,含 6 个 EXPECT_OK + 12 个 EXPECT_RAISE 用例。每次校验器改动,以及在 CI 中,都要在 skill 目录下跑一遍:

```bash
# 在项目根目录
python3 skills/pyspark-sql-guardrails/scripts/test_sql_guard.py

# 或在 scripts/ 目录下
cd skills/pyspark-sql-guardrails/scripts
python3 test_sql_guard.py
```

预期输出:6 行 `[OK] EXPECT_OK`、12 行 `[OK] EXPECT_RAISE: raised as expected`,以 `ALL TESTS PASSED` 收尾。出现任何偏差都说明校验器有回归 — 修校验器,不是修测试。

### 第 4 步 — 失败处理协议

当校验器抛错(在 CLI 形式下就是 `FAIL - <reason>` 这一行):

1. **停下流水线**,不要继续到下一阶段
2. **读 reason**,它会指明被违反的规则(空 / 多语句 / 错的 verb / 禁用关键字 / WITH 后没 SELECT)
3. **不要**为了放过这条 SQL 而去改校验器。校验器是权威,SQL 是错的
4. **不要**用正则剥除禁用关键字来"清理"SQL。要拒绝,要么重新生成 SQL
5. **重跑校验器**用修好的 SQL,循环到 `PASS`

### 禁止行为(校验器执行相关)

| 自我说服 | 现实 |
|---|---|
| "我就 `spark.sql(sql)`,信它" | 这正是防护栏要拦的。跑校验器。 |
| "校验器的输出在上一轮里" | 校验器只有通过函数调用才"有状态",再跑一遍。 |
| "就一行,肉眼看看就够了" | 肉眼判断正是这个防护栏要防的失败模式。 |
| "我已经对一条类似的 SQL 跑过了" | 每条 SQL 字符串都是新值,要针对实际要交付的字符串重跑。 |
| "我直接内联一段正则,不用 `validate_sql.py`" | skill 只给一个权威 CLI。内联副本会漂移、错过更新。 |
| "SQL 在代码块里,还没'执行'" | 把 SQL 展示给用户就已经是执行了。关口在代码块之前,不在之后。 |

## 常见自我说服(Common Rationalizations)

| 借口 | 现实 |
|---|---|
| "只是元数据刷新而已" | `REFRESH`、`MSCK`、`ANALYZE` 都不在 SELECT/INSERT 范围,先问。 |
| "MERGE 是非破坏性的" | 策略只允许 SELECT/INSERT,`MERGE` 一律禁止。 |
| "DELETE+INSERT 是已有模式" | 已有不安全模式不能凌驾于规则之上。 |
| "配置是可信的" | 配置里的 SQL 在 `spark.sql` 前一样要过白名单。 |
| "我可以用正则剥除禁用词" | 不要把不安全的 SQL 改写成看似安全的样子,直接拒绝。 |
| "我们要在 insert 之前做清理" | 用 SELECT 派生干净数据,再用 INSERT 写入;**不要**变更/删除已有数据。 |
| **"SQL 是硬编码的,不用校验。"** | **硬编码字符串一样要走 `assert_select_or_insert`。该函数检查的是字符串本身,不是它的出处。源码里的字面量并不比配置里的值更安全。** |
| **"我肉眼已经看过了,verb 是 SELECT。"** | **肉眼判断正是这个防护栏要防的失败模式。信函数,别信眼睛。`assert_select_or_insert` 没跑过,SQL 在定义上就是不可信的。** |
| **"我到下一轮/下一个 PR 再包。"** | **不行,现在就在调用点加上 wrapper。已提交代码里裸 `spark.sql(...)` 是回归,不是 TODO。** |

## 红旗信号(Red Flags)

看到下面任一项就要停下,加校验或先问用户:

- `spark.sql(config["sql"])`、YAML SQL、env SQL、CLI SQL,或来自外部输入的 f-string SQL
- 任何不是 SELECT/WITH/INSERT 的 SQL verb
- `INSERT OVERWRITE` 没拿到针对 overwrite 语义的明确批准
- 用分号分批的 SQL
- 分区修复、schema 迁移、过期分区清理、合并压实、vacuum、统计信息收集的请求
- 措辞为"快速清理一下"、"就刷一下元数据"、"删掉旧分区"、"复用 DELETE+INSERT" 的请求
- **直接的 `spark.sql(sql)` 调用,前面没有 `assert_select_or_insert(sql)`(也没用 `safe_spark_sql`)— 即便 SQL 字符串是硬编码字面量**
- **某条 PR、diff 或代码评审引入了新的 `spark.sql(...)` 调用点,却没加 wrapper**

## 常见错误(Common Mistakes)

- 只检查 `DELETE` 和 `TRUNCATE`;Spark 的风险还包括 `DROP`、`ALTER`、`MERGE`、`UPDATE`、`MSCK`、`REFRESH`、`ANALYZE`、`VACUUM`、`OPTIMIZE`
- 只校验了生成的 SQL,没校验配置里加载的 SQL
- 因为首条语句是 SELECT,就允许了多语句堆叠
- 把注释当作无害,但后面跟着一句禁用的 SQL
- 在 SQL 里用 `CREATE OR REPLACE TEMP VIEW`;如果需要临时视图,改用 DataFrame 的 `createOrReplaceTempView`
- **直接调 `spark.sql(sql)`,不走 `assert_select_or_insert` / `safe_spark_sql`。白名单在调用点生效,不在 SQL 生成时生效。一条"看起来"安全的 SQL,在函数跑过它之前都不算"已校验"安全。**
- **写内联的 `if sql.startswith("SELECT"): spark.sql(sql)`。这不是校验,只有 `assert_select_or_insert`(或等价且白名单一致的 AST 解析器)才算。**

---

## 在流水线中的位置(Pipeline Position)

本 skill 是 PySpark SQL 流水线的**写入/执行关口**:

```
requirements-analysis → metadata-validator → logic-planner → sql-context-builder → pyspark-sql-guardrails → sql-review
```

**强依赖的上游 skill:** `sql-context-builder` — 每条生成的 SQL 必须带有一份固化了 alias、join key 和字段来源的 `sql_context`。没有伴随 context 的 SQL 在 `sql-review` 中会命中 `HIGH` 风险(见 check #1:reference integrity)。

**强依赖的下游 skill:** `sql-review` — 本防护栏校验"允许哪些语句";`sql-review` 校验"这条被允许的语句是否正确"。两者都通过之后才能 `spark.sql(...)`。**不要**因为防护栏过了就跳过 `sql-review`。

**执行关口(本 skill 强制):** SQL 字符串通往集群的唯一通道是 `assert_select_or_insert(sql)`(或其封装 `safe_spark_sql(spark, sql)`)。直接的 `spark.sql(sql)` 是流水线违规。这条规则凌驾于流水线顺序 — 即便 `sql-review` 已经通过,执行时仍要过这一道防护栏。防护栏与评审是相互独立的检查,一道过不能代替另一道。

**本防护栏不检查:**

- 列是否真的存在于表(那是 `sql-review` 通过 SQL Context 校验的事)
- join key 是否正确(那是 `sql-review` 的事)
- 聚合是否被重复、行是否会膨胀(那是 `sql-review` 的事)
- 是否带时间/分区过滤(那是 `sql-review` 的事)

**本防护栏检查:**

- 第一个真实关键字是 `SELECT`、`WITH ... SELECT` 或 `INSERT ... SELECT`
- 没有 `;` 堆叠的多语句
- 体内不出现任何禁用的 verb
- `INSERT OVERWRITE` 必须基于用户的明确批准

校验器抛错时,改 SQL,**不要**削弱校验器。拿不准时,在生成 `INSERT OVERWRITE` 或任何 DDL 之前先问用户。