特征转换管道

生成有用的特征通常不仅仅涉及简单的数据选取。它经常需要应用一系列操作，将原始输入数据转换为适合机器学习 (machine learning)模型的表示形式。虽然临时转换脚本可能足以满足初步实验需求，但构建可扩展和可维护的ML系统需要更结构化的方法。这种结构化方法体现在高级特征存储中特征转换管道的定义、管理和运行。

将转换从分散的脚本或应用程序代码转移到与特征定义直接关联的受控管道中，具有显著益处：

1. 一致性： 确保在训练数据生成（离线）和推理 (inference)请求（在线）期间应用完全相同的逻辑，减少训练-服务偏差的常见来源。
2. 可重用性： 明确定义的转换步骤可以在不同特征组甚至项目间重用，减少重复工作。
3. 可发现性与治理： 将转换逻辑与特征定义一起集中管理，使其更容易理解特征如何派生、追踪其血缘并应用治理策略。
4. 可维护性： 转换逻辑的更新可以被系统地管理、版本化和部署。
5. 自动化： 有助于集成到MLOps工作流程中，用于自动化特征计算和部署。

定义转换管道

在特征存储的背景下，转换管道表示应用于输入数据源以生成一个或多个输出特征的有序操作序列，通常是一个有向无环图（DAG）。这些操作可以从简单的数据清洗和类型转换，到复杂的统计计算或预训练 (pre-training)模型的应用（如嵌入 (embedding)查找）。

常见的转换步骤包括：

• 数据清洗： 处理缺失值（填充），修正异常值，标准化格式。
• 编码： 将分类变量转换为数值表示（例如，独热编码、目标编码、标签编码）。
• 缩放/归一化 (normalization)： 将数值特征调整到共同的尺度（例如，StandardScaler，MinMaxScaler）。
• 特征交叉： 创建现有特征之间的交互项。
• 数学函数： 应用对数、多项式或三角函数。
• 日期/时间提取： 派生组件，如星期几、一天中的小时或时间差。
• 文本处理： 分词 (tokenization)，TF-IDF计算，应用嵌入模型。
• 自定义逻辑： 通过用户定义函数（UDF）实现特定领域的计算。

这种管道的定义通常发生在特征存储的框架或SDK内部。您不仅指定原始数据源，还要定义要应用的一系列转换。

# 示例：使用SDK定义管道

from feature_store_sdk import FeatureView, Transformation, source, registry
from my_custom_transforms import calculate_risk_score # 示例用户定义函数 (UDF)

# 定义数据源
user_activity_source = source(
    name="user_activity_stream",
    source_type="kafka",
    topic="user-events",
    event_timestamp_column="event_ts"
)

# 定义单个转换
impute_session_duration = Transformation(
    name="impute_duration_mean",
    function="mean", # 可以是内置函数或引用库函数
    inputs=["session_duration_raw"],
    outputs=["session_duration_imputed"],
    params={"default_value": 0} # 如果初始无法计算均值，则为默认值
)

scale_interaction_count = Transformation(
    name="scale_interactions_standard",
    function="standard_scaler", # 引用标准缩放实现
    inputs=["interaction_count"],
    outputs=["interaction_count_scaled"]
)

calculate_custom_score = Transformation(
    name="calculate_user_risk",
    function=calculate_risk_score, # 引用自定义Python函数
    inputs=["interaction_count_scaled", "session_duration_imputed"],
    outputs=["user_risk_score"]
)

# 定义特征视图，并关联管道
user_engagement_features = FeatureView(
    name="user_engagement_v1",
    entities=["user_id"],
    source=user_activity_source,
    # 管道被定义为一个转换列表/DAG
    pipeline=[
        impute_session_duration,
        scale_interaction_count,
        calculate_custom_score # 依赖于前一步骤的输出
    ],
    ttl="30d",
    online=True,
    offline=True
)

# 注册特征视图（包括其管道）
registry.apply(user_engagement_features)

这种声明式方法使得特征存储系统能够理解血缘（即 user_risk_score 如何依赖于 interaction_count_scaled 和 session_duration_imputed，而它们又依赖于原始输入）并管理执行。

集成与执行模型

这些已定义的管道如何执行取决于特征存储的架构和具体情况：

1. 批处理摄取/回填： 为了填充离线存储或回填历史数据，特征存储通常会协调一个分布式处理任务（例如，使用Spark、Flink或其自己的引擎）。定义的管道DAG被转换为应用于大型批处理数据集的可执行任务。结果（最终特征）被写入离线存储（例如，数据湖、数据仓库）。
2. 流式摄取： 对于实时特征，管道通常部署在流处理引擎中（例如，Flink、Kafka Streams、Spark Streaming）。当新事件从源流到达时，它们会通过管道中定义的转换步骤，并将生成的特征写入在线存储（可能也写入离线存储）。对于像滚动窗口聚合这样的转换，状态管理在此变得重要。
3. 按需计算： 一些特征存储支持在请求时计算特征。在这种情况下，检索存储的管道定义，并使用请求中提供的输入执行必要的转换，可能与从在线存储获取的数据结合。由于延迟限制，这对于复杂管道来说不太常见，但对于特定用例可能有用（第2.5节会涉及）。

集成程度有所不同。一些特征存储直接使用内置运算符或通过嵌入 (embedding)执行内核（如Python解释器或JVM）来执行管道逻辑。其他则采用更松散耦合的方法，特征存储管理定义和编排，但依赖外部计算引擎（如专用Spark集群或无服务器函数）来实际运行转换。选择会影响操作复杂度、成本和性能特点。

管道结构可视化

可视化通常有助于理解转换管道中的依赖关系。一个简单的DAG可以展示数据和操作的流程。

一个有向无环图（DAG），表示特征转换管道示例中的依赖关系。原始数据经过填充和缩放步骤后，用于自定义风险评分计算，最终填充特征视图。

管理管道复杂性和可重用性

随着特征和转换数量的增加，有效管理这些管道变得非常重要。高级特征存储提供以下机制：

• 版本控制： 转换逻辑会演进。管道，就像特征本身一样，需要版本控制以确保可重现性并允许安全地推出更改。将特定特征视图版本与特定转换版本关联是标准做法。
• 模块化和可重用性： 将常见的转换步骤（如特定指标的公司标准缩放器）定义为可重用组件，可以导入到多个管道中，从而减少代码重复并确保一致性。
• 测试： 转换逻辑应可独立进行单元测试。此外，特征存储环境中的集成测试是必要的，以验证管道在批处理和流处理上下文 (context)中的执行，并检查数据类型不匹配或意外空值等问题。
• 依赖管理： 特征存储的元数据层和执行引擎负责明确定义和追踪由不同管道或同一管道内转换生成的特征之间的依赖关系。

通过将转换逻辑嵌入 (embedding)到特征存储内的受控管道中，您可以创建更易于维护且一致的特征工程流程。这种结构化方法是扩展机器学习 (machine learning)操作和构建可靠生产系统的根本，为处理更复杂的场景（如流式特征和基于时间的聚合）做准备，我们将在下一部分进行讨论。