findscripter 何时使用
- 为数据管道/数仓建立可重复、可报警的质量校验：用 Great Expectations 写期望套件、用 dbt test 写表/列级测试、用数据契约约束上下游 schema 与 SLA。
- 把校验接入 CI/CD 或调度（Airflow/dbt Cloud），让脏数据在进入下游前被拦截，失败时报警（Slack 等）。
- 监控质量指标：完整性、唯一性、有效性、准确性、一致性、新鲜度。
- 触发词：数据质量、data quality、Great Expectations/GE、dbt test、数据契约/data contract、校验规则、完整性、唯一性、新鲜度。
不该用：
- 单机离线 CSV/Excel 表的清洗（去重、缺失填充、类型规整）→ 用
csv-data-cleaner。 - 纯查询/联表/聚合建模本身 → 用
sql-query-builder（本技能只校验其产物质量）。 - 表外数据（API/文件原貌）的 schema 推断与解析。
步骤 / 指令
- 定维度：先按六维（完整性/唯一性/有效性/准确性/一致性/新鲜度）圈定要测的关键列，不要全表全列乱测。
- 分层测（数据测试金字塔，自下而上）：
- schema 测试（结构）：列集合、列类型、行数区间。
- 单列单元测试：非空、唯一、取值集合、数值/日期区间。
- 跨表集成测试：外键关系、跨列约束、A>B 之类一致性。
- 选工具：仓内 dbt 优先用
dbt test（声明在_models.yml，配合dbt_utils）；需要丰富断言/统计期望/数据文档用 Great Expectations；跨团队边界用数据契约（datacontract.com + SodaCL）固化 schema 与 SLA。 - 落新鲜度：必测
created_at之类的时间列在近 N 天/小时内，陈旧数据等于坏数据。 - 编排+报警：用 GE checkpoint 或 dbt 在 CI/调度中批量跑，任一关键表失败即让管道失败并发通知。
- 演进：契约/schema 版本化，发现新问题就增量补测，阈值用动态基线而非硬编码。
示例
GE 安装与最小套件（核心命令保留自源）：
pip install great_expectations
great_expectations init
great_expectations datasource new
import great_expectations as gx
context = gx.get_context()
suite = context.add_expectation_suite("orders_suite")
suite.add_expectation(gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id"))
suite.add_expectation(gx.expectations.ExpectColumnValuesToBeUnique(column="order_id"))
# 取值集合 / 区间 / 新鲜度 / 行数 等同理：
# ExpectColumnValuesToBeInSet(column="status", value_set=[...])
# ExpectColumnValuesToBeBetween(column="amount", min_value=0, strict_min=True)
# ExpectColumnMaxToBeBetween(column="created_at", min_value="now-1d", max_value="now")
result = context.run_checkpoint(checkpoint_name="orders_checkpoint")
if not result.success:
raise ValueError("Data quality checks failed!")
dbt 声明式测试（models/marts/.../_models.yml）：
models:
- name: fct_orders
tests:
- dbt_utils.recency: {datepart: day, field: created_at, interval: 1} # 新鲜度
columns:
- name: order_id
tests: [unique, not_null]
- name: customer_id
tests:
- not_null
- relationships: {to: ref('dim_customers'), field: customer_id} # 外键
- name: order_status
tests:
- accepted_values: {values: [pending, processing, shipped, delivered, cancelled]}
- name: total_amount
tests:
- dbt_utils.expression_is_true: {expression: ">= 0"}
自定义通用 dbt 测试（行数区间）：
{% test row_count_in_range(model, min_count, max_count) %}
with row_count as (select count(*) as cnt from {{ model }})
select cnt from row_count where cnt < {{ min_count }} or cnt > {{ max_count }}
{% endtest %}
数据契约骨架（datacontract.com + SodaCL 质量块）：
apiVersion: datacontract.com/v1.0.0
kind: DataContract
metadata: {name: orders, version: 1.0.0, owner: data-platform-team}
schema:
properties:
order_id: {type: string, format: uuid, required: true, unique: true}
customer_id: {type: string, required: true, pii: true}
status: {type: string, enum: [pending, processing, shipped, delivered, cancelled]}
quality:
type: SodaCL
specification:
checks for orders:
- missing_count(order_id) = 0
- duplicate_count(order_id) = 0
- freshness(created_at) < 24h
sla: {availability: 99.9%, freshness: 1 hour}
注意事项
- 聚焦关键列，别测一切；过度测试拖慢管道又难维护。
- 早测、增量测：变换前先校验源数据；遇到问题再补测，并为每个测试写清晰描述。
- 别忽略 warning，它常是 failure 前兆；别跳过新鲜度校验。
- 阈值用动态基线，勿硬编码；行数/均值类统计期望要随业务波动调整。
- 测关系而非孤立列：外键、跨表一致性比单列断言更能抓真实坏账。
- 失败必报警（接监控/Slack），并让 CI/调度真正失败而非仅记录。
- 契约/schema 版本化，schema 变更可追踪；PII 列在契约中标注且不得下泄到下游 mart。
互见
- requires：无。
- related：
csv-data-cleaner（单机离线表的清洗，与入库后的质量校验互补）；sql-query-builder（被校验对象多由 SQL 建模产出）。 - combines_with：无。
本条采编自 wshobson/agents（MIT）。