验证模式和类型

数据管道失效通常不是因为转换逻辑有误，而是因为输入假设被违背了。当上游API将整数ID变为UUID字符串，或日期格式从 YYYY-MM-DD 转变到 MM-DD-YYYY 时，依赖特定内存布局或解析逻辑的下游进程会崩溃。这种现象称为模式漂移。

为避免这种情况，我们将模式不单单视为文档，更视为一份可强制执行的约定。在生产环境中，这份约定必须以程序方式验证，在数据获准进入转换层之前。

显式约定的作用

Python等动态语言中，数据类型通常在运行时推断。这虽有利于快速开发，却给数据工程带来不小的风险。包含 ["1", "2", "3"] 的列可能被推断为字符串对象，而 [1, 2, 3] 则被推断为整数。如果数据管道预期整数以执行连接操作，字符串推断将导致静默故障或笛卡尔积爆炸，具体取决于所使用的SQL方言或数据框库。

我们将模式 $S$ 定义为一组约束，它将列名 $C$ 映射到特定的数据类型 $T$。

$S = \{ (c_i, t_i) \mid c_i \in C, t_i \in T \}$

当一批新的数据 $D$ 到达时，验证函数 $f(D, S)$ 必须仅在 $D$ 中的每条记录都满足 $S$ 中的类型约束时才返回 True。

在Pandas中实现模式验证

一种常见的验证方法是将传入数据框的 dtypes 与定义的预期字典进行比较。但是，标准的相等检查不足够，因为它们未考虑可空类型或特定的整数精度（例如 int64 对 int32）。

您必须进行一项细致的类型检查，既验证是否存在，也验证类型是否兼容。

import pandas as pd
import numpy as np

def validate_schema(df: pd.DataFrame, expected_schema: dict) -> bool:
    """
    验证数据框的列是否与预期类型匹配。
    """
    errors = []
    
    # 1. 检查缺失列
    missing_cols = set(expected_schema.keys()) - set(df.columns)
    if missing_cols:
        errors.append(f"Missing columns: {missing_cols}")
    
    # 2. 检查现有列的类型
    for col, expected_type in expected_schema.items():
        if col in df.columns:
            actual_type = df[col].dtype
            # 如果架构上允许，可以有一定的灵活性（例如，int64 对 int32）
            if not np.issubdtype(actual_type, expected_type):
                errors.append(
                    f"Column '{col}' type mismatch. "
                    f"Expected subtype of {expected_type}, got {actual_type}"
                )

    if errors:
        for e in errors:
            print(f"Schema Error: {e}")
        return False
        
    return True

# 使用示例
expected = {
    'user_id': np.integer,
    'revenue': np.floating,
    'timestamp': np.datetime64
}

# 模拟一个错误批次（ID为字符串）
data = {
    'user_id': ['101', '102'], 
    'revenue': [50.25, 100.00],
    'timestamp': pd.to_datetime(['2023-01-01', '2023-01-02'])
}

df_batch = pd.DataFrame(data)

# 这将失败，因为 user_id 是对象/字符串，而不是整数
is_valid = validate_schema(df_batch, expected)

在上面的示例中，验证起到了一个关卡的作用。np.issubdtype 的使用允许进行分层类型检查。例如，要求 np.integer 将接受 int32 或 int64，这在拒绝字符串或浮点数的同时提供了必要的灵活性。

验证关卡的架构

验证应尽可能早地在数据管道中进行，通常是在提取（ELT）或加载（ETL）之后立即进行。这能防止“污染”数据与数据湖或数据仓库中的可信资产混合。

下图展现了模式验证器在摄取工作流中的位置。

验证关卡根据模式合规性路由数据，将格式错误记录隔离到死信队列（DLQ）以供人工检查。

处理可空性与类型强制转换

模式错误的常见来源涉及空值。在PostgreSQL或Snowflake等标准数据库系统中，列被定义为 NULLABLE（可空）或 NOT NULL（非空）。在Pandas 1.0之前的版本中，整数不能包含 NaN 值；它们会自动被强制转换为浮点数。

如果您的模式预期一个整数ID，但源数据发送了一个空值，整个列可能会被强制转换为浮点数（101 变为 101.0）。这在与其他整数键进行连接时会破坏参照完整性。

为应对这种情况，工程师必须在验证步骤之前强制执行显式类型转换逻辑，或使用感知空值的类型（例如 Pandas 的 Int64）。

$f_{cast}(x) = \begin{cases} \text{整数}(x) & \text{若 x 非空} \\ \text{错误} & \text{若 x 为空且模式要求细致} \end{cases}$

管理嵌套结构

Parquet和Avro等现代数据格式支持复杂的嵌套类型（结构体和数组）。验证扁平模式很简单，但验证JSON列需要递归逻辑。

验证嵌套结构时，您有两种策略：

1. 写入时模式（细致）: 您使用Pydantic等库或JSON Schema定义来规定JSON数据块的精确结构。如果出现额外字段，或嵌套字段类型不正确，则该记录会被拒绝。
2. 读取时模式（演进式）: 您只验证路由或识别所需的顶级键。JSON数据块的其余部分作为通用 variant 或 struct 类型存储，留待后续解析。

对于可靠的生产数据管道，对于核心业务指标，推荐采用细致方法；而演进式方法适用于日志或功能标志等预期会有模式漂移的场景。

类型分布的可视化

统计分析有助于展现大型数据集中的类型一致性。即使某一列在技术上存储为“字符串”，它也可能包含混合的语义类型（例如，“123”、“N/A”、“456”）。下图展现了一个列的特征，该列在技术上作为字符串是有效的，但在语义上不一致。

对列进行分析表明，虽然存储类型一致（字符串），但内容混合了整数、UUID和无效文本。

在这种情况下，一个简单的 df.dtypes 检查会通过，因为所有内容都是字符串。验证工具需要更进一步，使用正则表达式来验证字符串内容是否符合预期格式（例如，^\d+$ 用于数字字符串）。