用AI和Gherkin革新数据质量

AI模型价格对比 | AI工具导航 | ONNX模型库 | Vibe Coding教程 | Tripo 3D | Meshy AI | ElevenLabs | KlingAI | ArtSpace | Phot.AI | InVideo

想象一下：你刚刚完成了一个包含20列的复杂DBT模型构建。接下来就是繁琐的部分——为每一列编写测试，检查空值、验证唯一性、确保引用完整性，以及覆盖边界情况。

如果你和大多数数据工程师一样，你会写几个基本测试就收工了。也许你会给ID列加上not_null和unique测试，但其余19列呢？业务逻辑验证呢？边界情况呢？

这就是我六个月前在做一个数据管道项目时面临的现实。我们的团队构建了50多个DBT模型，但测试覆盖率只有30%左右。我们知道需要更好的数据质量保障，但编写全面的测试既耗时又枯燥。

就在那时，我有了一个想法：如果我们能用通俗的英文描述测试需求，让AI自动生成全面的、生产就绪的DBT测试呢？

1、解决方案

我构建的工具结合了三个革命性的概念：

1.1 Gherkin DSL：行为的语言

Gherkin是一种用于行为驱动开发（BDD）的领域特定语言。它的设计目标是让人类可读、让机器可执行。以下是它为什么非常适合数据测试的原因：

Feature: Customer Data Quality
  Ensure customer data meets business standards

Scenario: Customer ID validation
  Given a customer table with customer_id column
  When we check for data quality issues
  Then customer_id should be unique
  And customer_id should not be null
  And customer_id should match pattern 'CUST-[0-9]+'

注意到它多么易读吗？即使是非技术人员也能理解我们在测试什么。这成为了我们的规范语言。

1.2 AI驱动的生成：从意图到实现

我没有手动将Gherkin规范转换为DBT测试，而是集成了大语言模型（LLM）来完成这项繁重的工作。系统支持：

• OpenAI的GPT-4用于复杂的测试生成
• Anthropic的Claude用于细腻的理解
• Azure OpenAI用于企业级部署
• 本地LLM（通过Ollama）用于隔离环境

AI不仅仅生成基本测试——它理解上下文、推断关系，并创建覆盖12个不同测试类别的全面测试套件。

1.3 全面覆盖：100%的目标

传统测试生成器为每列创建一两个测试。我的工具平均为每列生成3个以上的测试，覆盖：

• 空值检查：必填字段验证
• 唯一性：主键和唯一键验证
• 数据类型验证：类型转换检查
• 格式验证：邮箱、电话、URL模式
• 范围验证：日期和数值边界
• 业务逻辑：跨字段验证规则
• 引用完整性：外键关系
• 枚举验证：可接受的值列表
• 长度验证：字符串长度约束
• 数值验证：数值范围检查
• 日期验证：时间逻辑验证
• 边界情况：极端值和空白字符

2、架构：它如何工作

让我带你了解系统架构：

2.1 输入层：三种表达意图的方式

方式1：纯英文

I need to validate a customer table that has unique customer IDs,
properly formatted email addresses, and status values of only
'active', 'inactive', or 'suspended'

AI会自动将其转换为结构化的Gherkin。

方式2：Gherkin规范

Feature: Order Processing Quality
  Validate order data integrity

Scenario: Order status validation
  Given an orders table with status column
  Then status should have accepted_values ['pending', 'completed', 'cancelled']

喜欢结构化方式的用户可以直接编写Gherkin。

方式3：GitHub/本地模型

# Load existing DBT models
models = load_from_github('https://github.com/org/repo/models')

# Generate tests automatically
generate_tests(models)

系统会分析现有SQL并生成相应的测试。

2.2 处理层：AI驱动的测试生成

这里是魔法发生的地方。系统：

1. 分析输入（纯英文、Gherkin或SQL）
2. 提取列信息和数据类型
3. 推断关系和约束
4. 生成使用模板和AI的全面测试SQL
5. 验证生成的代码是否兼容DBT

AI提示词工程至关重要。以下是一个简化版本：

def generate_comprehensive_tests(model_name, columns):
    prompt = f"""
    Generate comprehensive DBT tests for model: {model_name}
    Columns: {columns}
    
    Include:
    - Null checks for all columns
    - Uniqueness for ID columns
    - Format validation for email/phone/URL columns
    - Range checks for date/numeric columns
    - Business logic validation
    - Edge case handling
    
    Output valid DBT SQL with proper config blocks.
    """
    
    return llm.generate(prompt)

2.3 输出层：生产就绪的产物

系统生成五个关键输出：

1. schema.yml：列级测试（unique、not_null、accepted_values）
2. 单元测试SQL：全面的行为测试
3. DBT模型：实际的转换逻辑
4. 覆盖率报告：详细的测试覆盖率分析
5. 文档：使用指南和最佳实践

3、结果：实际影响

在生产环境中部署该工具后，我们看到了显著的改善：

之前：

• 平均测试覆盖率：30%
• 编写测试时间：每个模型2-3小时
• 测试维护：高（脆弱、不一致）
• Bug检测率：40%（生产环境中）

之后：

• 平均测试覆盖率：100%
• 生成测试时间：每个模型30秒
• 测试维护：低（标准化、全面）
• Bug检测率：85%（生产前）

关键指标

• 98%的减少在测试编写时间上
• 3.3倍的增长在测试覆盖率上
• 2.1倍的提升在Bug检测上
• **每年节省$50K+**在数据质量事故成本上

4、实现细节深入

让我分享一些让这个工具成功的关键实现细节：

4.1 智能列分析

系统使用模式匹配来智能地应用测试：

def analyze_column(column_name, column_type):
    tests = ['null_check']  # All columns get this
    
    if 'id' in column_name.lower():
        tests.extend(['uniqueness', 'positive_integer'])
    
    if 'email' in column_name.lower():
        tests.append('email_format')
    
    if 'date' in column_name.lower() or 'timestamp' in column_name.lower():
        tests.extend(['date_range', 'not_future', 'not_ancient'])
    
    if 'status' in column_name.lower() or 'type' in column_name.lower():
        tests.append('enum_validation')
    
    return tests

4.2 模型类型支持

不同的DBT物化方式需要不同的方法：

增量模型：

select * from {{ source('raw', 'events') }}
{% if is_incremental() %}
  where event_timestamp > (select max(event_timestamp) from {{ this }})
{% endif %}

快照模型（Type 2 SCD）：

{% snapshot orders_snapshot %}
{{
    config(
      target_schema='snapshots',
      unique_key='order_id',
      strategy='timestamp',
      updated_at='updated_at'
    )
}}
select * from {{ source('raw', 'orders') }}
{% endsnapshot %}

临时模型（CTE）：

{{ config(materialized='ephemeral') }}
-- Compiled as CTE in downstream models
select id, clean_data from {{ source('raw', 'staging') }}

4.3 测试覆盖率矩阵

可视化测试覆盖率矩阵成为了最受欢迎的功能：

Column        | Null | Unique | Type | Format | Range | Business | ...
------------- | ---- | ------ | ---- | ------ | ----- | -------- | ---
customer_id   |  ✅  |   ✅   |  ✅  |   ➖   |  ➖   |    ➖    | ...
email         |  ✅  |   ➖   |  ✅  |   ✅   |  ➖   |    ➖    | ...
created_at    |  ✅  |   ➖   |  ✅  |   ➖   |  ✅   |    ✅    | ...
status        |  ✅  |   ➖   |  ✅  |   ➖   |  ➖   |    ✅    | ...

这种可视化帮助团队快速识别覆盖率的空白。

5、我学到的最佳实践

5.1 从纯英文开始

不要强迫你的团队立即学习Gherkin语法。让他们自然地描述需求：

"我们的客户ID应该是唯一的，永远不能为空。邮箱地址需要有效。状态只能是active或inactive。"

AI会处理转换。

5.2 审查AI生成的测试

AI很强大但并不完美。在部署之前始终审查生成的测试：

-- AI might generate this:
where email not like '%@%'

-- But you might want this:
where email !~ '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}$'

5.3 结合传统和AI生成

对标准测试使用传统的基于规则的生成，对复杂场景使用AI：

# Rule-based for standard tests
generate_standard_tests(model)

# AI for complex business logic
generate_ai_tests(complex_requirements)

5.4 对Gherkin规范进行版本控制

将Gherkin规范当作代码来对待：

/tests
  /specs
    customer_validation.feature
    order_processing.feature
  /generated
    test_customers.sql
    test_orders.sql

这使得测试的演进变得可追溯。

6、挑战与解决方案

挑战1：LLM幻觉

问题： AI有时会生成无效的SQL或不存在的DBT函数。

解决方案： 实现验证层：

def validate_generated_sql(sql):
    # Check for valid DBT syntax
    if '{{ ref(' not in sql and '{{ source(' not in sql:
        raise ValidationError("Missing DBT references")
    
    # Verify test structure
    if 'config(' not in sql:
        raise ValidationError("Missing config block")
    
    # Ensure returns fail condition
    if 'where' not in sql.lower():
        raise ValidationError("Test must filter for failures")
    
    return sql

挑战2：上下文窗口限制

问题： 大型模型超出LLM上下文窗口。

解决方案： 分块处理：

def process_large_model(model_sql):
    chunks = split_into_chunks(model_sql, max_tokens=3000)
    
    tests = []
    for chunk in chunks:
        tests.extend(generate_tests(chunk))
    
    return merge_and_deduplicate(tests)

挑战3：测试执行时间

问题： 100%覆盖率意味着很多测试，这可能会拖慢CI/CD。

解决方案： 并行执行和智能测试选择：

# Run only changed model tests
dbt test --select state:modified+

# Parallel execution
dbt test --threads 8

# Severity-based filtering
dbt test --severity error  # Run critical tests only

7、未来：下一步计划

我目前正在开发三个主要增强功能：

7.1 异常检测集成

将传统测试与基于ML的异常检测相结合：

# Traditional test
assert no_nulls(customer_id)

# ML-based anomaly detection
detect_distribution_shift(revenue, threshold=0.05)
detect_outliers(transaction_amount, method='isolation_forest')

7.2 自愈式测试

当测试失败时，自动生成假设和修复方案：

if test_fails(test_result):
    hypothesis = ai_diagnose(test_result, model_sql)
    fix = ai_suggest_fix(hypothesis)
    
    if user_approves(fix):
        apply_fix(fix)
        regenerate_tests()

7.3 自然语言查询

用自然语言提问关于测试覆盖率的：

"哪些模型的覆盖率低于80%？" "哪些列最经常缺乏测试？" "显示上周所有失败的测试"

8、结束语

构建这个工具让我认识到，数据质量的未来不在于编写更多的测试——而在于描述"好的"是什么样的，然后让AI来处理实现。

原文链接：Revolutionizing Data Quality: Building AI-Powered DBT Tests with Gherkin DSL

汇智网翻译整理，转载请标明出处