ml-data-pipeline

机器学习数据管道

概述

在数据摄取、验证、预处理、特征工程、数据集版本管理、特征存储、批处理和流式处理管道以及数据质量监控中使用此技能。在机器学习中，数据管道的正确性往往比模型的复杂程度更重要。管道必须生成可避免数据泄漏的训练数据，以及一致的服务特征。

数据管道不变量

• 原始数据不可变，或可通过快照寻址。
• 训练和服务前必须验证架构、统计信息和质量预期。
• 转换必须进行版本管理，并且可复现。
• 必须在过采样、目标编码、特征选择或归一化等容易引入数据泄漏的操作之前创建数据集拆分。
• 时间相关特征只能使用预测时可用的信息。
• 离线训练特征必须与在线服务特征一致。
• 敏感数据应尽可能减少，并进行访问控制、加密和审计。

摄取和存储

根据数据形态和访问模式选择存储方式。对于表格批处理机器学习，使用 Parquet/Arrow 的对象存储是一个不错的默认选择。Delta Lake、Apache Iceberg 或 Hudi 可提供 ACID 表、架构演进和时间旅行功能。使用数据仓库存储受治理的 SQL 特征，使用向量存储进行嵌入检索，并使用流式日志记录在线行为。分别存储原始、清洗后、特征和模型就绪的数据层。对于 ADF 用于传递 Azure ML 代码资产版本的 Azure Storage 指针 Blob，请加载 ml-azureml-adf-automation。

对于大型数据集，优先使用列式格式、按时间或高层级领域进行分区、压缩、谓词下推和清单文件。避免产生大量微小文件；必要时进行压缩合并。记录每次训练运行的数据集快照标识符。

Feast 特征存储蓝图

Feast 是一个模块化特征存储，用于维护在线与离线特征的一致性。

1. 特征存储配置（feature_store.yaml）

project: fraud_detection
registry: data/registry.db
provider: local
online_store:
  type: redis
  connection_string: "localhost:6379"
offline_store:
  type: file

2. 特征定义（features.py）

from datetime import timedelta
from feast import Entity, FeatureView, Field, FileSource, ValueType
from feast.types import Float32, Int64

user = Entity(name="user_id", value_type=ValueType.INT64, join_keys=["user_id"])

user_transactions_source = FileSource(
    path="data/user_transactions.parquet",
    event_timestamp_column="timestamp",
    created_timestamp_column="created_timestamp",
)

user_transactions_fv = FeatureView(
    name="user_transactions_feature_view",
    entities=[user],
    ttl=timedelta(days=90),
    schema=[
        Field(name="transaction_count_30d", dtype=Int64),
        Field(name="total_amount_30d", dtype=Float32),
    ],
    online=True,
    source=user_transactions_source,
)

处理引擎

引擎	最适用场景
pandas	小型到中型的内存中探索和简单管道
Polars	快速的本地／惰性列式处理，以及处理规模超过 pandas 的工作负载
Spark	大规模分布式 ETL、湖仓工作流、SQL 与 MLlib 集成
Dask	Python 原生的分布式数组／数据框和自定义工作负载
Ray Data	以机器学习为中心的分布式预处理，并与 Ray Train/Tune/Serve 集成
Beam/Flink/Spark Streaming	流式或统一的批处理／流式数据流
Airflow/Prefect/Dagster	编排、调度、重试、血缘和依赖管理

高性能 Polars 惰性聚合管道

Polars 惰性求值通过谓词下推和投影下推来优化执行计划。

import polars as pl

def compute_rolling_user_features(transactions_path: str):
    lazy_df = (
        pl.scan_parquet(transactions_path)
        # Cast timestamp for windowing
        .with_columns(pl.col("timestamp").str.strptime(pl.Datetime))
        .sort("timestamp")
        # Define rolling calculation window
        .group_by_dynamic(
            index_column="timestamp",
            every="1d",
            period="30d",
            group_by="user_id"
        )
        .agg([
            pl.col("amount").count().alias("transaction_count_30d"),
            pl.col("amount").sum().alias("total_amount_30d"),
            pl.col("amount").mean().alias("avg_amount_30d")
        ])
        .filter(pl.col("user_id").is_not_null())
    )
    # Collect executes the query optimization and loads results into memory
    return lazy_df.collect()

Spark Dataframe 优化方案

避免数据倾斜和过多 Shuffle 开销等常见的分布式训练瓶颈。

1. 使用加盐防止连接倾斜

from pyspark.sql import functions as F

# Adding a salt column to distribute skewed key values evenly across partitions
skewed_df = skewed_df.with_columns(
    (F.rand() * 10).cast("int").alias("salt")
)
lookup_df = lookup_df.with_columns(
    F.explode(F.array([F.lit(i) for i in range(10)])).alias("salt")
)

# Join on key AND salt to distribute the workload
joined_df = skewed_df.join(lookup_df, ["join_key", "salt"], "inner").drop("salt")

2. 对查找表使用广播连接

from pyspark.sql.functions import broadcast

# Explicitly broadcast small dimension dataframe to executors to avoid shuffling large fact table
optimized_joined = large_fact_df.join(broadcast(small_lookup_df), "entity_id", "inner")

数据验证

在摄取、特征生成、训练和服务阶段进行验证。检查架构、类型、范围、空值、唯一性、重复项、类别取值范围、基数、标签分布、时间戳单调性、引用完整性、文本／图像／音频有效性以及嵌入范数。可用工具包括 Great Expectations、TensorFlow Data Validation、Deequ、pandera、dbt tests 和自定义断言。

对于契约违规，质量检查应立即失败；对于需要调查的分布变化，则发出警告。将验证报告与训练产物一同存储。在生产环境中，同时监控原始特征和转换后的模型输入。

特征工程

特征工程应与预测时点紧密关联。对于时序数据，计算滚动窗口时应采用正确的截止点和延迟，并妥善处理延迟到达的数据。对于目标编码，应在交叉验证的各折内拟合编码器，并使用平滑处理。对于类别特征，应根据基数和模型类型，选择原生类别支持、独热编码、哈希、嵌入或目标编码。对于文本，应对分词器和词表进行版本管理。对于图像／音频，应保存预处理参数和数据增强方式。

要防止数据泄漏，可以这样问：模型做出预测时，这个特征是否已知？如果未知，就应将其排除，或重新设计目标和预测时点。

DVC（数据版本控制）摄取工作流

DVC 通过轻量级元数据文件，将大型数据集关联到 Git 提交，避免仓库膨胀。

1. 初始化并添加存储

dvc init
dvc remote add -d myremote s3://my-dvc-bucket/raw-data

2. 跟踪新的数据集版本

# Add dataset to DVC tracking (creates data.parquet.dvc)
dvc add data/raw_transactions.parquet

# Commit DVC metadata file to Git
git add data/raw_transactions.parquet.dvc data/.gitignore
git commit -m "Track transactions dataset v1.0.0 via DVC"

# Push raw binaries to remote cloud storage
dvc push

3. 在另一台工作机上检索已跟踪的版本

git pull
dvc pull

流式和在线流水线

流式机器学习流水线需要处理事件时间、水位线、去重、排序策略、迟到数据行为、恰好一次或至少一次语义，以及可重放性。应将在线特征更新与训练标签生成分开。特征逻辑变更时，应能通过持久化日志进行回填或重放。

安全与隐私

尽量减少敏感字段，并在适当情况下对标识符进行令牌化或哈希处理；只有在需要时才保留关联能力。按数据层实施访问控制。避免将原始个人身份信息（PII）导出到实验跟踪工具或模型制品中。记录数据保留和删除政策。

常见的流水线故障

• 由于预处理实现不一致，导致训练与服务阶段出现偏差。
• 滚动特征、聚合或目标编码中出现未来信息泄漏。
• 静默的模式漂移导致字符串、空值或类别被错误转换。
• 训练集与验证集中包含重复实体。
• 在数据集拆分前进行过采样或归一化。
• 数据集未进行版本管理，导致训练无法复现。
• 流式特征无法回填到训练数据中。

资料来源

• Great Expectations 文档：https://docs.greatexpectations.io/
• TensorFlow Data Validation 文档：https://www.tensorflow.org/tfx/data_validation
• Feast 文档：https://docs.feast.dev/
• DVC 文档：https://dvc.org/doc
• Apache Spark MLlib 和 SQL 文档：https://spark.apache.org/docs/latest/
• Polars 文档：https://docs.pola.rs/