Snowflake 数据验证

2025年7月30日 | 阅读 14 分钟

在你的数据管道中，数据验证需要成为一个常驻居民，而不是仅仅一个过客。Snowflake 为数据管道增加了两个新且引人注目的功能——动态表以及告警和通知。借助 Snowflake 的动态表，你可以使用 SQL 语句、子查询、聚合、窗口函数和 Python 函数（包括 Snowpark）轻松构建可靠的数据管道。

这个强大的工具也可以用于数据验证，无论表中的数据是移动的还是静态的。让我们举一个例子来更好地解释这个想法。虽然 Snowflake 的快速入门指南有一个详尽的动态表快速入门指南，但我们将通过展示一个数据验证用例来提供预览。

假设你是一家互联网零售商的数据工程师或数据治理开发人员。你的目标是确保业务团队能及时收到库存预警，以便他们能够补货低库存产品。你定期从一个事实表中生成一个详细的 SALESREPORT。此表记录了产品的详细信息和相应的销售数量。

同时，你的产品表包含了截至上次补货日期的重要库存信息。在生成我们的 SALESREPORT 事实表后，你将在每次运行时使用这两个表进行一些简单的 SQL 查询。

我们的主动方法是通过我们的数据收集来确定某个产品的库存是否低于 10%。让我们在 Snowflake 中一个新的“数据验证动态表” SQL 工作表中实现这一点。

 
USE SCHEMA DEMO.DT_DEMO;
CREATE OR REPLACE DYNAMIC TABLE PROD_INV_ALERT
    LAG = '1 MINUTE'
    WAREHOUSE=LAB_S_WH
AS
    SELECT 
        S.PRODUCT_ID, 
        S.PRODUCT_NAME,CREATION TIME AS LATEST_SALES_DATE,
        STOCK AS BEGINING_STOCK,
        SUM(S.QUANTITY) OVER (PARTITION BY S.PRODUCT_ID ORDER BY CREATION TIME) TOTALUNITSOLD, 
        (STOCK - TOTALUNITSOLD) AS UNITS LEFT,
        ROUND(((STOCK-TOTALUNITSOLD)/STOCK) *100,2) PERCENT_UNITLEFT,
        CURRENT_TIMESTAMP() AS ROWCREATIONTIME
    FROM SALES REPORT S JOIN PROD_STOCK_INV ON PRODUCT_ID = PID
    QUALIFY ROW_NUMBER() OVER (PARTITION BY PRODUCT_ID ORDER BY CREATION TIME DESC) = 1;   

现在让我们检查一下是否有任何产品的库存不足

 
-- Check products with low inventory and alert
select * from prod_inv_alert where percent_unitleft < 10;   

你可以使用告警和通知来帮助你的产品采购和库存人员通过发送电子邮件告警来补充必要的商品。别忘了在下面的代码中更新仓库和电子邮件地址。

 
CREATE NOTIFICATION INTEGRATION IF NOT EXISTS
    notification_emailer
    TYPE=EMAIL
    ENABLED=TRUE
    ALLOWED_RECIPIENTS=('first.last@company.com')
    COMMENT = 'email integration to update on low product inventory levels'
;

CREATE OR REPLACE ALERT alert_low_inv
  WAREHOUSE = my_warehouse
  SCHEDULE = '30 MINUTE'
  IF (EXISTS (
      SELECT *
      FROM prod_inv_alert
      WHERE percent_unitleft < 10 and ROWCREATIONTIME > SNOWFLAKE.ALERT.LAST_SUCCESSFUL_SCHEDULED_TIME()
  ))
  THEN CALL SYSTEM$SEND_EMAIL(
                'notification_emailer', -- notification integration to use
                'first.last@company.com', -- Email
                'Email Alert: Low Inventory of products,' -- Subject
                'Inventory running low for certain products. Please check the inventory report in Snowflake table prod_inv_alert' -- Body of email.
);

-- Alerts are paused by default, so let's resume it first
ALTER ALERT alert_low_inv RESUME;

-- Add new records
insert into sales data select * from table(gen_cust_purchase(10000,2));   

只有当新信息添加到动态数据库（低库存产品）时，这些通知才会激活。因此，在 Snowflake 中维护和管理实时数据的告警非常简单。

这些告警可以暂停、恢复和监控。

 
-- Monitor alerts in detail
SHOW ALERTS;
SELECT *
FROM
  TABLE(INFORMATION_SCHEMA.ALERT_HISTORY(
 SCHEDULED_TIME_RANGE_START
      =>dateadd('hour',-1,current_timestamp())))
WHERE
    NAME = 'ALERT_LOW_INV'
ORDER BY SCHEDULED_TIME DESC;
-- Suspend Alerts 
-- Important step to suspend alert and stop consuming the warehouse credit
ALTER ALERT alert_low_inv SUSPEND;   

Snowflake 云数据仓库迁移

从主要基于云的 Snowflake 事实数据仓库迁移到传统的 Netezza 信息数据仓库涉及几个步骤。数据验证是数据迁移项目成功的关键。在数据迁移的每个阶段以及端到端数据验证情况下的数据间隙，都可以利用 DataFlow 来验证数据。

步骤 1：从遗留数据仓库提取数据

数据通常在提取为 CSV 或 Parquet 格式后传输到 AWS S3 登陆区域。AWS 提供了几种将文件传输到 S3 的选项，具体取决于数据量。

在数据发送到 AWS S3 后，必须进行数据验证，以确保所有数据都已正确检索并传输到 AWS S3。这些测试通常是 AWS S3 登陆区域中的文件数据与遗留数据仓库中的表数据的逐一比较，因为在此步骤中转换不多。

比较表中的行数与文件中的行数。
比较数据编码。
比较数据完整性。
比较数据值。

步骤 2：数据转换

测试用例图示例如右所示。可以通过 JDBC 组件读取遗留数据仓库。可以使用文件组件从 AWS S3 读取数据。最后，可以使用数据比较组件比较这两个数据集。下面是数据比较组件输出的示例。

在将数据加载到 Snowflake 之前，还可以使用数据整理来提高数据质量。为了在整理数据之前发现数据问题，对数据进行剖析并进行数据质量测试至关重要。这些任务可以通过 DataFlow 来完成。

使用数据规则和数据剖析组件来发现数据质量问题。
比较登陆区域和 S3 中的暂存（已整理）区域之间的数据。
处理数据整理并与暂存区域同步

步骤 3：将数据复制到 Snowflake

假设 Snowflake 表已建立，复制数据到 Snowflake 是最后一步。使用 VALIDATE 函数来验证数据文件并查找任何错误。加载后，可以使用 DataFlow 来比较 Snowflake 和暂存区域（S3）文件之间的数据。

比较表与文件行数
比较数据编码
比较数据完整性
比较数据值

如右图所示，可以使用 DataFlow 在一个测试中完成端到端数据验证。可以使用单个 DataFlow 比较遗留数据仓库和 S3 的数据，以及遗留数据仓库和 Snowflake 的数据。

步骤 4：通过修改 Snowflake 来修改报表

完成这些调整后，必须进行严格的测试程序，将遗留数据仓库生成的报表与其对应的 Snowflake 生成的报表进行比较。

比较报表的数据和布局。
比较报表的性能。
通过模拟并发用户负载，在新的环境中对报表进行压力测试。
检查安全性。

使用 Datagaps BI Validator（一种无代码 BI 测试解决方案）可以自动执行支持的 BI 工具的所有这些测试。有必要验证是否所有数据都已有效迁移。许多哈希函数——包括 Snowflake 中提供的函数（例如 Oracle 中的 ORA_HASH、Snowflake 中的 HASH 等）——被用来尝试验证迁移，但哈希输出不匹配。本教程提供了将数据迁移到 Snowflake 以验证数据的方法。

由于不同数据库中的方法和函数是专有的，使用不同的方法创建，并且产生的哈希结果与 Snowflake 中的哈希函数不同，因此无法使用其 HASH 等效函数来比较不同数据库中的数据完整性验证。迁移后验证来自不同数据库的数据完整性（表中的行）的解决方案是使用非数据库特定的函数，如 MD5 校验和，该函数将产生相同的哈希输出。

示例

如果行中没有 null 值

select seq,first_name,last_name, md5(concat(seq,first_name,last_name)) as hash

from my stable;

SEQ FIRST_NAME LAST_NAME HASH
Craig AB 4724218126181ffb5c4ee9db630d5cd3
zxxzx newuser 1a673ebe945628f0f23e8bfa43a99a83
fourth last fourth user 61f25d4c974b7b74c5c66816f8aa41cb

如果行中有 null 列，则使用 coalesce

 
select seq, first_name,last_name, md5(concat(seq,coalesce(first_name,'a'),coalesce(last_name,'a')) ) as hash
from my stable;   

Snowflake 数据质量验证

有效利用数据需要信任和高质量的数据。借助 DataBuck，Snowflake 用户可以通过为每个数据资产分配信任分数来评估其数据质量。DataBuck 通过自动检测每个数据集独特的数据质量，将发现、调查和创建数据验证标准所需的工作量减少了 95%。

从手动模型转向基于信任的自动化策略。

DataBuck 利用其机器学习功能，为每个数据资产（模式、表和列）确定客观的数据信任分数。数据驱动的信任将不再基于受欢迎程度。无需让人们对表或文件的质量进行主观评分。客观的数据信任分数被所有各方广泛理解。

工作原理

扫描：DataBuck 扫描 Snowflake 平台上的每个数据资产。每次调度程序调用 DataBuck 或数据资产刷新时，都会重新扫描资产。数据不会传输到 DataBuck，而是就地进行扫描。
自动发现指标：DataBuck 会自动生成数据健康指标，这些指标对于每个数据资产都是独特的。使用 AI/ML 算法，公认且标准化的 DQ 测试专门针对每块数据进行定制。
监控：为了识别不可接受的数据风险，根据每个数据资产中各列的质量特征计算健康指标，并随时间推移进行跟踪。数据信任分数是从健康指标派生出来的。
告警：当信任分数降至特定阈值以下时，DataBuck 会通知用户。它通过持续监控健康指标和信任分数来实现这一点。

数据质量的维度级别

结果摘要中显示了信任分数的偏差。它显示了用户可以信任数据的程度以及自上次两次研究以来质量和健康状况的变化。

有关更多详细信息，请双击找到的任何违规项

用户可以放大维度，以查看哪些列受到数据资产级别的at影响。要查看列的特定维度，请单击其名称。有关更多信息，请单击列级别的维度名称。之后，用户可以确定特定的数据质量违规是应被忽略还是应报告进行进一步检查，无论是针对整个数据资产还是单个列。

为了保证数据质量和完整性，可以在 Snowflake 数据模型中应用数据验证约束和检查。数据验证过程可防止不准确或不一致的数据进入数据库，并有助于强制执行业务标准。可以使用以下方法将限制和数据验证检查合并到 Snowflake 中

1. 检查限制

Snowflake 支持检查约束，即在数据被添加到表或在表中修改之前必须评估为真的条件。可以对特定列或一组列应用检查约束。它们有助于根据预定标准实现数据验证规则。

示例

 
Create a table with a check constraint.
CREATE TABLE employees (
employee_id INT PRIMARY KEY,
salary NUMERIC(10, 2),
hire_date DATE,
CONSTRAINT salary_check CHECK (salary > 0)
);   

2. NOT NULL 约束

使用 NOT NULL 约束来强制要求特定列必须具有非空值。这确保在数据插入期间始终提供必要的数据。

3. UNIQUE 限制

UNIQUE 要求规定，指定列中的值在整个表中必须是唯一的。这有助于维护数据完整性并避免重复条目。

示例

 
Create a table with UNIQUE constraint.
CREATE TABLE products (
product_id INT PRIMARY KEY,
product_name VARCHAR,
product_code VARCHAR UNIQUE
);   

4. 外键约束

外键约束通过确保一个表中的数据与另一个表中的数据匹配来维护参照完整性。它们强制表之间的关系并防止孤立记录。

5. 正则表达式 (REGEXP)

您可以使用正则表达式来验证和强制执行文本数据中的特定模式。

6. 用户定义函数 (UDFs)

Snowflake 允许您设计 UDFs 来根据业务逻辑执行复杂的检查和个性化数据验证。

7. 物化视图

预先聚合数据和根据预定义标准执行数据验证检查是物化视图的两种用途。它们在提高查询性能的同时保持数据质量。

通过将这些数据验证检查和约束合并到 Snowflake 数据模型中，您可以保证数据质量，维护数据完整性，并强制执行业务规则。通过采取这些预防措施，您可以阻止不准确的数据被添加，并提高数据的整体质量和可靠性。

确保导入到 Snowflake 或从 Snowflake 查询的数据是准确、一致并符合必要的数据完整性标准或业务规则，这就是 Snowflake 数据验证。以下是在 Snowflake 中验证数据的通用方法

1.模式验证

数据类型验证：在 Snowflake 中，表中的每列都被分配了一个特定的数据类型。导入数据到 Snowflake 时，确保传入数据与指定的数据类型匹配至关重要。
可空性验证：任何输入到指定为 NOT NULL 的列中的数据都不应包含 NULL 值。您可以通过执行 SQL 查询来查看这些列中是否存在任何意外的 NULL 值。
约束检查：即使 Snowflake 不在数据库级别强制执行主键和外键约束，您也可以设置检查来确保数据符合这些约束。例如，您可以检查应唯一的列是否存在重复值。

2. 数据完整性验证

行数验证：加载后，应将 Snowflake 表中的行数与源数据中的行数进行比较。这是一个简单但重要的检查。行数差异可能是由于加载不完整或数据丢失等问题。
数据完整性：确保所有必需的列都已填写。例如，关键列（如订单 ID）不应出现 NULL 或缺失的条目。
范围检查：某些列中的值应仅在特定范围内。例如，“折扣”列应仅包含 0 到 100 之间的数字。您可以创建 SQL 查询来查找任何超出预期范围的数字。

3. 业务规则验证

特定业务规则：您的组织的独特要求可能决定了特殊业务规则的必要性。例如，金融数据集中的交易应平衡（贷方应等于借方）。可以使用 SQL 查询来验证这些规则。
摘要和聚合：可能需要将预期值与聚合数据（如总销售额）进行比较。任何显著的差异都可能表明数据加载或转换过程中存在问题。

4. 数据质量验证

重复检查：查找重复记录至关重要，特别是在记录（如客户数据库）应是唯一的表中。使用关键列标准，您可以创建查询来识别重复项。
异常值检测：当合法数据偏离预期范围时，识别和检查这些异常值至关重要。例如，涉及异常大或小的销售交易可能需要进一步研究。
检查模式：验证数据是否遵循特定模式。例如，确保所有电话号码格式正确，所有电子邮件地址都符合 xxx@yyy.zzz 的格式。

5. ETL/ELT 流程验证

转换验证：在 ETL/ELT 操作期间，数据经常被修改（例如，聚合数据，更改数据格式）。验证这些修改是否正确应用至关重要。您可以将数据转换前后的数据进行比较，以确保数据准确。
加载验证：验证所有数据是否已加载，并且没有数据被跳过或重复加载。可以通过查看日志、验证行数和与源数据进行比较来验证此项。
审计日志：大多数 ETL 解决方案（包括 Snowflake）都提供可用于审计数据加载的日志。检查这些日志有助于查找警告或加载失败等问题。

6. Snowflake 特有的验证工具

Snowflake 查询：数据验证可以直接在 Snowflake 环境中进行，利用 Snowflake 提供的强大 SQL 引擎。可以开发查询来验证数据的准确性、完整性和一致性。
存储过程：您可以在 Snowflake 中创建和运行存储过程，这有助于自动化复杂的验证操作。可以设置存储过程以按计划或响应特定事件执行。
第三方技术：许多第三方技术与 Snowflake 集成，提供高级数据验证功能。通过 Informatica、Talend 和 dbt（数据构建工具）等工具可以实现复杂的验证过程、检查自动化和报告生成。

7. 性能验证

查询性能：验证查询是否高效执行，尤其是在处理大型数据集时。性能不佳的查询可能是由于不正确的聚类、缺少索引或低效的查询架构等问题。
索引/聚类：为了最大化数据检索，Snowflake 使用聚类键。为了提高查询效率，请确保您的表应用了正确的聚类键。

8. 性能和一致性评估

查询性能验证
1. 方法：监控查询执行时间，确保它们符合性能标准。
2. 工具：使用 Snowflake 的分析和查询历史记录工具来检查性能。
3. 最佳方法：建立性能阈值，并经常评估缓慢的查询以进行优化。
跨环境的数据一致性
1. 方法：确保数据在开发、暂存和生产环境之间保持一致。
2. 工具：使用脚本或自动化工具来比较环境之间的行数、列数据和关键指标。

9. 自动化验证和测试

单元测试
1. 工具：使用 dbt（数据构建工具）等框架编写特定数据验证的测试。
2. 示例：例如，验证列是否无空值或特定值是否唯一。
测试
1. not_null
  1. column_name
2. unique
  1. column_name
回归测试
1. 工具：使用 dbt（数据构建工具）等框架编写特定数据验证的测试。
2. 示例：例如，验证列是否无空值或特定值是否唯一。
3. 回归测试方法：在对数据管道进行修改后，验证当前数据验证是否仍然通过。

可以使用 Snowflake Tasks 和 Streams（一种调度和监控工具）来安排基于 SQL 的数据验证检查。

10. 错误处理和记录

错误日志记录程序：为将来的检查记录和存储验证错误。
工具：使用 Snowflake 的日志记录功能或第三方工具（如 Apache Airflow 和 dbt）来管理日志。
最佳方法：将日志保存在一个中央位置（例如，一个特殊的 Snowflake 表），以便于访问。设置告警以通知团队验证失败是告警方法。
工具：连接到 PagerDuty、Snowflake 的事件驱动告警或自定义电子邮件告警等监控平台。