Great Expectations 数据验证框架4 个对象把坏数据拦在管道外【免费下载链接】great_expectationsAlways know what to expect from your data.项目地址: https://gitcode.com/GitHub_Trending/gr/great_expectations凌晨两点训练任务挂了排查半天发现是一行 passenger_count0 的脏数据混进了出租车样本。这类问题靠人眼巡检拦不住Great Expectations 这个开源数据验证框架的思路是把数据应该长什么样写成期望Expectation让每批数据先过一遍检查再放行。全程 Python 驱动跑完还附赠可视化报告。一条坏数据如何走到报告里框架里真正干活的就四类对象各管一段Data Context整个项目的入口配置、期望套件、验证结果都存在它下面Datasource / Batch Request告诉框架数据在哪、这一轮检查哪一批Expectation Suite规则集合乘客数必须在 1 到 6 之间就是其中一条Checkpoint把取一批数据 → 跑一遍规则 → 出结果固化成可复跑的流程。坏数据的旅程就是反向的它先被 Batch Request 圈进检查范围被某条期望拦下最终以失败行的形式出现在报告里。两条命令装好 Great ExpectationsPython 环境建议 3.8~3.11先建虚拟环境再装包python -m venv gx_env source gx_env/bin/activate pip install great-expectations需要连接数据库或 Spark 时用 extras 把驱动带上pip install great-expectations[spark,sqlalchemy,pandas]⚠️ 各数据库还有自己的驱动依赖比如 PostgreSQL 要另装psycopg2-binarySnowflake 要装snowflake-connector-python缺了会在建连接时报错而不是在安装时报错。数据上下文给验证立一个本地项目Data Context 分内存Ephemeral、文件File和云Cloud三种形态本地开发用文件型最合适它会把项目落到磁盘上跨会话也不丢from great_expectations.data_context import FileDataContext context FileDataContext.create(project_root_dir./gx_project)这一行会生成标准目录great_expectations.yml是主配置expectations/存规则集合data_docs/存报告。实现代码在 great_expectations/data_context/出问题时可以顺着找。把 CSV 注册成数据源第一组期望怎么定义数据源配置只需说清两件事执行引擎用谁、文件在哪。以./data下的出租车 CSV 为例datasource_config { name: taxi_data, class_name: Datasource, execution_engine: {class_name: PandasExecutionEngine}, data_connectors: { # 扫描 ./data 下所有 .csv文件名去扩展名即资产名 default_inferred_data_connector_name: { class_name: InferredAssetFilesystemDataConnector, base_directory: ./data, default_regex: {pattern: (.*)\\.csv, group_names: [data_asset_name]}, } }, } context.add_datasource(**datasource_config)拿到数据源后用 Batch Request 指到具体那张表再开一个 Validator 来写规则batch_request context.get_batch_request( datasource_nametaxi_data, data_connector_namedefault_inferred_data_connector_name, data_asset_nameyellow_tripdata, ) context.create_expectation_suite(taxi_data_suite, overwrite_existingTrue) validator context.get_validator(batch_requestbatch_request, expectation_suite_nametaxi_data_suite) validator.expect_column_values_to_be_between(passenger_count, 1, 6, mostly0.95) validator.expect_column_values_to_not_be_null(fare_amount) validator.expect_table_columns_to_match_ordered_list( [passenger_count, trip_distance, fare_amount] ) validator.save_expectation_suite(discard_failed_expectationsFalse)mostly0.95是关键参数允许 5% 的异常值规则不会因个别脏行整批翻红。想卡类型就加一条expect_column_values_to_be_of_type比如 trip_distance 是 float64。所有规则按前缀分三层expect_table_*管表级列名、行数expect_column_*管单列expect_column_pair_*管列间关系选的时候先问一句这条规则作用于几列。检查点把期望变成可复跑的流水线Checkpoint 解决的是这套规则如何反复执行。把套件和 Batch Request 绑到检查点上再用 action_list 声明跑完之后做什么——存结果、刷新文档、发通知都可以挂在这里context.add_checkpoint( nametaxi_checkpoint, class_nameSimpleCheckpoint, expectation_suite_nametaxi_data_suite, batch_requestbatch_request, action_list[ {name: store, action: {class_name: StoreValidationResultAction}}, {name: docs, action: {class_name: UpdateDataDocsAction}}, ], ) result context.run_checkpoint(checkpoint_nametaxi_checkpoint) print(✅ 通过 if result[success] else ❌ 失败)success这个布尔值是接 CI 的钩子流水线里只看它一个字段红灯直接阻断。检查点的实现见 great_expectations/checkpoint/。验证报告失败行如何被定位context.build_data_docs()这一步在data_docs/下生成 HTML 站点不用每次手写报告。报告的价值在失败定位每条期望单独一行通过的打勾失败的会给出10000 行里有 1579 个未预期值15.79%这样的观测值并附上采样到的坏行排查时不用再跑一遍 SQL。接入命令行与 Airflow 调度不想写 Python 脚本时用 CLI检查点配置和执行都是标准命令great_expectations checkpoint new taxi_checkpoint great_expectations checkpoint run taxi_checkpoint挂到 Airflow 上就是包一层 BashOperator按天调度from airflow import DAG from airflow.operators.bash import BashOperator from datetime import datetime with DAG(taxi_data_validation, default_args{owner: data_team, start_date: datetime(2024, 1, 1)}, schedule_intervaldaily) as dag: validate BashOperator( task_idvalidate_taxi_data, bash_commandgreat_expectations checkpoint run taxi_checkpoint, )生产环境可以再往 action_list 里加 Slack、邮件通知把失败推送给值班的人。安装与数据源连接常见问题现象处理No module named great_expectations虚拟环境没激活激活后重装ImportError: Pandas requires version...依赖版本冲突把 pandas 锁到兼容版本Permission denied用虚拟环境或pip install --userPostgreSQL 连不上装psycopg2-binary执行引擎换成SqlAlchemyExecutionEngine用connection_stringpostgresql://user:passhost:port/db传连接信息 检查点能在 CI 里稳定跑起来之后下一步值得做两件事用 Profiler 从数据自动推断第一版期望省掉逐条手写把同一份套件共享给上下游当作数据契约来用。【免费下载链接】great_expectationsAlways know what to expect from your data.项目地址: https://gitcode.com/GitHub_Trending/gr/great_expectations创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Great Expectations 数据验证框架：4 个对象把坏数据拦在管道外
Great Expectations 数据验证框架4 个对象把坏数据拦在管道外【免费下载链接】great_expectationsAlways know what to expect from your data.项目地址: https://gitcode.com/GitHub_Trending/gr/great_expectations凌晨两点训练任务挂了排查半天发现是一行 passenger_count0 的脏数据混进了出租车样本。这类问题靠人眼巡检拦不住Great Expectations 这个开源数据验证框架的思路是把数据应该长什么样写成期望Expectation让每批数据先过一遍检查再放行。全程 Python 驱动跑完还附赠可视化报告。一条坏数据如何走到报告里框架里真正干活的就四类对象各管一段Data Context整个项目的入口配置、期望套件、验证结果都存在它下面Datasource / Batch Request告诉框架数据在哪、这一轮检查哪一批Expectation Suite规则集合乘客数必须在 1 到 6 之间就是其中一条Checkpoint把取一批数据 → 跑一遍规则 → 出结果固化成可复跑的流程。坏数据的旅程就是反向的它先被 Batch Request 圈进检查范围被某条期望拦下最终以失败行的形式出现在报告里。两条命令装好 Great ExpectationsPython 环境建议 3.8~3.11先建虚拟环境再装包python -m venv gx_env source gx_env/bin/activate pip install great-expectations需要连接数据库或 Spark 时用 extras 把驱动带上pip install great-expectations[spark,sqlalchemy,pandas]⚠️ 各数据库还有自己的驱动依赖比如 PostgreSQL 要另装psycopg2-binarySnowflake 要装snowflake-connector-python缺了会在建连接时报错而不是在安装时报错。数据上下文给验证立一个本地项目Data Context 分内存Ephemeral、文件File和云Cloud三种形态本地开发用文件型最合适它会把项目落到磁盘上跨会话也不丢from great_expectations.data_context import FileDataContext context FileDataContext.create(project_root_dir./gx_project)这一行会生成标准目录great_expectations.yml是主配置expectations/存规则集合data_docs/存报告。实现代码在 great_expectations/data_context/出问题时可以顺着找。把 CSV 注册成数据源第一组期望怎么定义数据源配置只需说清两件事执行引擎用谁、文件在哪。以./data下的出租车 CSV 为例datasource_config { name: taxi_data, class_name: Datasource, execution_engine: {class_name: PandasExecutionEngine}, data_connectors: { # 扫描 ./data 下所有 .csv文件名去扩展名即资产名 default_inferred_data_connector_name: { class_name: InferredAssetFilesystemDataConnector, base_directory: ./data, default_regex: {pattern: (.*)\\.csv, group_names: [data_asset_name]}, } }, } context.add_datasource(**datasource_config)拿到数据源后用 Batch Request 指到具体那张表再开一个 Validator 来写规则batch_request context.get_batch_request( datasource_nametaxi_data, data_connector_namedefault_inferred_data_connector_name, data_asset_nameyellow_tripdata, ) context.create_expectation_suite(taxi_data_suite, overwrite_existingTrue) validator context.get_validator(batch_requestbatch_request, expectation_suite_nametaxi_data_suite) validator.expect_column_values_to_be_between(passenger_count, 1, 6, mostly0.95) validator.expect_column_values_to_not_be_null(fare_amount) validator.expect_table_columns_to_match_ordered_list( [passenger_count, trip_distance, fare_amount] ) validator.save_expectation_suite(discard_failed_expectationsFalse)mostly0.95是关键参数允许 5% 的异常值规则不会因个别脏行整批翻红。想卡类型就加一条expect_column_values_to_be_of_type比如 trip_distance 是 float64。所有规则按前缀分三层expect_table_*管表级列名、行数expect_column_*管单列expect_column_pair_*管列间关系选的时候先问一句这条规则作用于几列。检查点把期望变成可复跑的流水线Checkpoint 解决的是这套规则如何反复执行。把套件和 Batch Request 绑到检查点上再用 action_list 声明跑完之后做什么——存结果、刷新文档、发通知都可以挂在这里context.add_checkpoint( nametaxi_checkpoint, class_nameSimpleCheckpoint, expectation_suite_nametaxi_data_suite, batch_requestbatch_request, action_list[ {name: store, action: {class_name: StoreValidationResultAction}}, {name: docs, action: {class_name: UpdateDataDocsAction}}, ], ) result context.run_checkpoint(checkpoint_nametaxi_checkpoint) print(✅ 通过 if result[success] else ❌ 失败)success这个布尔值是接 CI 的钩子流水线里只看它一个字段红灯直接阻断。检查点的实现见 great_expectations/checkpoint/。验证报告失败行如何被定位context.build_data_docs()这一步在data_docs/下生成 HTML 站点不用每次手写报告。报告的价值在失败定位每条期望单独一行通过的打勾失败的会给出10000 行里有 1579 个未预期值15.79%这样的观测值并附上采样到的坏行排查时不用再跑一遍 SQL。接入命令行与 Airflow 调度不想写 Python 脚本时用 CLI检查点配置和执行都是标准命令great_expectations checkpoint new taxi_checkpoint great_expectations checkpoint run taxi_checkpoint挂到 Airflow 上就是包一层 BashOperator按天调度from airflow import DAG from airflow.operators.bash import BashOperator from datetime import datetime with DAG(taxi_data_validation, default_args{owner: data_team, start_date: datetime(2024, 1, 1)}, schedule_intervaldaily) as dag: validate BashOperator( task_idvalidate_taxi_data, bash_commandgreat_expectations checkpoint run taxi_checkpoint, )生产环境可以再往 action_list 里加 Slack、邮件通知把失败推送给值班的人。安装与数据源连接常见问题现象处理No module named great_expectations虚拟环境没激活激活后重装ImportError: Pandas requires version...依赖版本冲突把 pandas 锁到兼容版本Permission denied用虚拟环境或pip install --userPostgreSQL 连不上装psycopg2-binary执行引擎换成SqlAlchemyExecutionEngine用connection_stringpostgresql://user:passhost:port/db传连接信息 检查点能在 CI 里稳定跑起来之后下一步值得做两件事用 Profiler 从数据自动推断第一版期望省掉逐条手写把同一份套件共享给上下游当作数据契约来用。【免费下载链接】great_expectationsAlways know what to expect from your data.项目地址: https://gitcode.com/GitHub_Trending/gr/great_expectations创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
