agents 仓库 contenteditable="false">【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents数据质量是数据管道可靠性的最后一道防线。本篇文章以 agents 仓库Multi-harness agentic plugin marketplace中data-engineering插件的># expectations/orders_suite.py import great_expectations as gx from great_expectations.core import ExpectationSuite from great_expectations.core.expectation_configuration import ExpectationConfiguration def build_orders_suite() - ExpectationSuite: Build comprehensive orders expectation suite suite ExpectationSuite(expectation_suite_nameorders_suite) # Schema expectations suite.add_expectation(ExpectationConfiguration( expectation_typeexpect_table_columns_to_match_set, kwargs{ column_set: [order_id, customer_id, amount, status, created_at], exact_match: False # Allow additional columns } )) # Primary key suite.add_expectation(ExpectationConfiguration( expectation_typeexpect_column_values_to_not_be_null, kwargs{column: order_id} )) suite.add_expectation(ExpectationConfiguration( expectation_typeexpect_column_values_to_be_unique, kwargs{column: order_id} )) # Foreign key suite.add_expectation(ExpectationConfiguration( expectation_typeexpect_column_values_to_not_be_null, kwargs{column: customer_id} )) # Categorical values suite.add_expectation(ExpectationConfiguration( expectation_typeexpect_column_values_to_be_in_set, kwargs{ column: status, value_set: [pending, processing, shipped, delivered, cancelled] } )) # Numeric ranges suite.add_expectation(ExpectationConfiguration( expectation_typeexpect_column_values_to_be_between, kwargs{ column: amount, min_value: 0, max_value: 100000, strict_min: True # amount 0 } )) # Date validity suite.add_expectation(ExpectationConfiguration( expectation_typeexpect_column_values_to_be_dateutil_parseable, kwargs{column: created_at} )) # Freshness - data should be recent suite.add_expectation(ExpectationConfiguration( expectation_typeexpect_column_max_to_be_between, kwargs{ column: created_at, min_value: {$PARAMETER: now - timedelta(days1)}, max_value: {$PARAMETER: now} } )) # Row count sanity suite.add_expectation(ExpectationConfiguration( expectation_typeexpect_table_row_count_to_be_between, kwargs{ min_value: 1000, # Expect at least 1000 rows max_value: 10000000 } )) # Statistical expectations suite.add_expectation(ExpectationConfiguration( expectation_typeexpect_column_mean_to_be_between, kwargs{ column: amount, min_value: 50, max_value: 500 } )) return suite逐条拆解其设计意图与参数含义表级结构期望expect_table_columns_to_match_set约束表必须包含的列集合exact_match: False表示允许未来新增列而不断言失败兼顾了模式演进。主键约束expect_column_values_to_not_be_nullexpect_column_values_to_be_unique组合等价于关系型数据库中的PRIMARY KEY语义。外键约束仅校验customer_id非空Great Expectations 属于单表引擎真正的跨表引用完整性需交给 Pattern 4 的 dbtrelationships测试或 Pattern 6 中跨表编排处理。类别有效性expect_column_values_to_be_in_set限定status只能出现在五个业务合法枚举中堵住脏枚举值。数值区间amount必须落在(0, 100000]strict_min: True将下界改为开区间即金额必须严格大于 0。日期合法性 新鲜度先校验created_at可被dateutil解析格式合法再用expect_column_max_to_be_between与$PARAMETER动态参数断言最大创建时间应落在过去一天到当前时刻之间从而捕获数据陈旧问题——这是 Timeliness 维度的典型落地。行数合理性expect_table_row_count_to_be_between用1000与10000000作为数量级的上下护栏防止全表丢失或数据爆炸。统计口径expect_column_mean_to_be_between断言订单均额处于50~500区间用于捕捉数值分布的整体漂移例如币种或单位换算错误。SKILL.md的快速上手部分展示了使用现代gx.expectations类式 API 的等价写法gx.expectations.ExpectColumnValuesToNotBeNull(columnorder_id)两种风格均可通过context.add_expectation_suite(...)登记使用。Pattern 2用 Checkpoint 把校验变成可调度的流水线仅定义 Suite 还不够Great Expectations 通过Checkpoint将数据批次 Expectation Suite 结果动作绑定成一个可独立运行的单元。details.md给出完整 YAML 配置# great_expectations/checkpoints/orders_checkpoint.yml name: orders_checkpoint config_version: 1.0 class_name: Checkpoint run_name_template: %Y%m%d-%H%M%S-orders-validation validations: - batch_request: datasource_name: warehouse data_connector_name: default_inferred_data_connector_name data_asset_name: orders data_connector_query: index: -1 # Latest batch expectation_suite_name: orders_suite action_list: - name: store_validation_result action: class_name: StoreValidationResultAction - name: store_evaluation_parameters action: class_name: StoreEvaluationParametersAction - name: update_data_docs action: class_name: UpdateDataDocsAction # Slack notification on failure - name: send_slack_notification action: class_name: SlackNotificationAction slack_webhook: ${SLACK_WEBHOOK} notify_on: failure renderer: module_name: great_expectations.render.renderer.slack_renderer class_name: SlackRenderer配置要点batch_request通过datasource_name、data_connector_name与data_asset_name定位数据源data_connector_query.index: -1明确表示只取最新一个批次保证每天调度校验的都是最新数据。expectation_suite_name与被校验批次绑定的 Suite即 Pattern 1 的orders_suite。action_list按顺序执行的结果动作链——StoreValidationResultAction持久化校验结果写入expectations存储StoreEvaluationParametersAction保存求值参数供下游数据文档引用UpdateDataDocsAction刷新 Data Docs 可视化报告SlackNotificationAction仅在失败notify_on: failure时通过${SLACK_WEBHOOK}环境变量注入的 Webhook 推送通知。随后给出通过 Python API 触发并做失败门禁的调用方式# Run checkpoint import great_expectations as gx context gx.get_context() result context.run_checkpoint(checkpoint_nameorders_checkpoint) if not result.success: failed_expectations [ r for r in result.run_results.values() if not r.success ] raise ValueError(fData quality check failed: {failed_expectations})注意result.run_results中的r是 Checkpoint 校验结果对象而非 Expectation 列表取到失败项后立即抛错可以让上层调度Airflow 任务、CI 步骤等感知失败——这与 Pattern 6 的校验失败即中断管道目标一致。初始化与创建数据源的命令行在SKILL.md中给出pip install great_expectations、great_expectations init、great_expectations datasource new。Pattern 3在 dbt 中用声明式 YAML 挂接数据测试当数据链路以 dbt 为核心时更自然的做法是直接在 model 的 schema 文件里声明测试。details.md给出了fct_orders事实表与dim_customers维表的完整测试声明# models/marts/core/_core__models.yml version: 2 models: - name: fct_orders description: Order fact table tests: # Table-level tests - dbt_utils.recency: datepart: day field: created_at interval: 1 - dbt_utils.at_least_one - dbt_utils.expression_is_true: expression: total_amount 0 columns: - name: order_id description: Primary key tests: - unique - not_null - name: customer_id description: Foreign key to dim_customers 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: - not_null - dbt_utils.expression_is_true: expression: 0 - name: created_at tests: - not_null - dbt_utils.expression_is_true: expression: current_timestamp - name: dim_customers columns: - name: customer_id tests: - unique - not_null - name: email tests: - unique - not_null # Custom regex test - dbt_utils.expression_is_true: expression: email ~ ^[A-Za-z0-9._%-][A-Za-z0-9.-]\\.[A-Za-z]{2,}$要点分析表级测试dbt_utils.recencycreated_at距当前最多滞后 1 天等效于新鲜度检查、dbt_utils.at_least_one表至少一行、dbt_utils.expression_is_truetotal_amount 0全表成立。列级测试unique/not_null覆盖主外键完整性relationships通过ref(dim_customers)将fct_orders.customer_id关联到dim_customers.customer_id这是 dbt 承担跨表集成测试数据测试金字塔顶层的机制accepted_values校验枚举集合email用正则表达式断言邮箱格式验证准确性维度。此处使用的stg_、int_、dim_/fct_分层命名与_core__models.yml的组织方式对应同一插件下 dbt-transformation-patterns 技能中描述的 medallion 架构sources → staging → intermediate → marts。Pattern 4自定义 dbt 测试generic 与 singular内置测试无法覆盖所有业务规则时dbt 允许自定义generic通用测试与singular单例测试。通用测试 1行数范围-- tests/generic/test_row_count_in_range.sql {% 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 %} -- Usage in schema.yml: -- tests: -- - row_count_in_range: -- min_count: 1000 -- max_count: 10000000通用测试通过 Jinja 宏接收model、min_count、max_count参数产出违反规则的行可被任意 model 复用等价于把 Great Expectations 的expect_table_row_count_to_be_between移植到 dbt 生态。通用测试 2序列值连续性-- tests/generic/test_sequential_values.sql {% test sequential_values(model, column_name, interval1) %} with lagged as ( select {{ column_name }}, lag({{ column_name }}) over (order by {{ column_name }}) as prev_value from {{ model }} ) select * from lagged where {{ column_name }} - prev_value ! {{ interval }} and prev_value is not null {% endtest %}使用窗口函数lag求相邻两行差值将与预设间隔不符的行全部选出用于流水号、序列主键等连续值场景。单例测试孤儿订单业务规则-- tests/singular/assert_orders_customers_match.sql -- Singular test: specific business rule with orders_customers as ( select distinct customer_id from {{ ref(fct_orders) }} ), dim_customers as ( select customer_id from {{ ref(dim_customers) }} ), orphaned_orders as ( select o.customer_id from orders_customers o left join dim_customers c using (customer_id) where c.customer_id is null ) select * from orphaned_orders -- Test passes if this returns 0 rowsSingular 测试用于一次性、强业务语义的断言此处为查出存在于订单表但不存在于客户维表的孤儿customer_id。dbt 的测试约定是返回 0 行即通过因此所有违反规则的样本行都会被当作失败输出天然自带可读的失败报告。Pattern 5用数据契约在团队边界锁定 schema 与质量承诺数据契约将 schema 定义、PII 标注与质量门槛文档化为可版本化、可评审的契约文件供上下游团队共同遵守。details.md给出基于 Data Contract Specificationdatacontract.com/v1.0.0的完整契约# contracts/orders_contract.yaml apiVersion: datacontract.com/v1.0.0 kind: DataContract metadata: name: orders version: 1.0.0 owner:># quality_pipeline.py from dataclasses import dataclass from typing import List, Dict, Any import great_expectations as gx from datetime import datetime dataclass class QualityResult: table: str passed: bool total_expectations: int failed_expectations: int details: List[Dict[str, Any]] timestamp: datetime class DataQualityPipeline: Orchestrate data quality checks across tables def __init__(self, context: gx.DataContext): self.context context self.results: List[QualityResult] [] def validate_table(self, table: str, suite: str) - QualityResult: Validate a single table against expectation suite checkpoint_config { name: f{table}_validation, config_version: 1.0, class_name: Checkpoint, validations: [{ batch_request: { datasource_name: warehouse, data_asset_name: table, }, expectation_suite_name: suite, }], } result self.context.run_checkpoint(**checkpoint_config) # Parse results validation_result list(result.run_results.values())[0] results validation_result.results failed [r for r in results if not r.success] return QualityResult( tabletable, passedresult.success, total_expectationslen(results), failed_expectationslen(failed), details[{ expectation: r.expectation_config.expectation_type, success: r.success, observed_value: r.result.get(observed_value), } for r in results], timestampdatetime.now() ) def run_all(self, tables: Dict[str, str]) - Dict[str, QualityResult]: Run validation for all tables results {} for table, suite in tables.items(): print(fValidating {table}...) results[table] self.validate_table(table, suite) return results def generate_report(self, results: Dict[str, QualityResult]) - str: Generate quality report report [# Data Quality Report, fGenerated: {datetime.now()}, ] total_passed sum(1 for r in results.values() if r.passed) total_tables len(results) report.append(f## Summary: {total_passed}/{total_tables} tables passed) report.append() for table, result in results.items(): status ✅ if result.passed else ❌ report.append(f### {status} {table}) report.append(f- Expectations: {result.total_expectations}) report.append(f- Failed: {result.failed_expectations}) if not result.passed: report.append(- Failed checks:) for detail in result.details: if not detail[success]: report.append(f - {detail[expectation]}: {detail[observed_value]}) report.append() return \n.join(report) # Usage context gx.get_context() pipeline DataQualityPipeline(context) tables_to_validate { orders: orders_suite, customers: customers_suite, products: products_suite, } results pipeline.run_all(tables_to_validate) report pipeline.generate_report(results) # Fail pipeline if any table failed if not all(r.passed for r in results.values()): print(report) raise ValueError(Data quality checks failed!)该管道的几个关键设计值得在生产中复用运行期动态构造 Checkpointvalidate_table用字典现场组装checkpoint_config并传给context.run_checkpoint(**checkpoint_config)免去为每张表手工维护一份 YAML结果归一化将 Great Expectations 的run_results展开成自有的QualityResultdataclass表名、是否通过、期望总数、失败数、明细与时间戳业务代码只需依赖这一个模型报告自动化generate_report产出 Markdown 格式报告汇总通过表数/总表数并只列出失败项及其observed_value方便直接粘贴进 Issue 或 IM失败即中断run_all结束后统一判断只要有任一张表未通过就打印报告并raise ValueError让 CI/CD 或 Airflow 能够据此终止管道避免坏数据继续向下游扩散。最佳实践Dos 与 DontsSKILL.md以工程实践清单收尾是任何落地者都应遵循的护栏应当做Dos尽早测试Test early——在转换transformation之前先校验源数据从源头拦截脏数据渐进式补测Test incrementally——在排查数据问题过程中持续追加测试让测试集随问题知识一起成长文档化每个期望Document expectations——为每条测试写清描述避免后人无法理解断言动机失败即告警Alert on failures——与监控体系Slack、PagerDuty 等集成版本化契约Version contracts——跟踪 schema 变更保证上下游同步演进。不要做Donts不要测试一切Dont test everything——聚焦关键列与关键业务规则测试也有维护成本不要忽视警告Dont ignore warnings——warnings 往往是失败的前兆不要跳过新鲜度检查Dont skip freshness——过期数据本身就是坏数据不要硬编码阈值Dont hardcode thresholds——用动态基线如$PARAMETER相对时间替代拍脑袋的固定值不要孤立地测试Dont test in isolation——同时测试表间关系如 dbtrelationships、孤儿数据检查单表测试通过不等于数据链路健康。仓库延伸技能在 plugins 生态中的组织方式想在本仓库进一步研究这套技能的落地方式可沿以下路径深入data-quality-frameworks/SKILL.md导航层包含数据质量维度表、测试金字塔、Great Expectations 快速上手与最佳实践清单data-quality-frameworks/references/details.md本文章核心来源六大 Pattern 的完整代码与配置data-engineer agent将该技能用于数据质量监控、告警与治理场景的 agent 定义其Response Approach第 4 步即在管道全链路加入数据质量检查与校验data-pipeline 命令 与 contenteditable="false">【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考