Skip to content
←

필사 모드: MLOps Feature Store 实战 — 用 Feast 构建特征流水线

中文
0%
정확도 0%
💡 왼쪽 원문을 읽으면서 오른쪽에 따라 써보세요. Tab 키로 힌트를 받을 수 있습니다.
MLOps Feature Store - Feast

概述

把 ML 模型部署到生产环境时，最常见的问题之一就是Training-Serving Skew(训练-服务偏差)。训练时使用的特征和服务时使用的特征不一致，导致模型性能下降的现象。Feature Store(特征存储)是从根本上解决这个问题的基础设施组件，它把特征的定义、存储、服务集中管理起来。

Feast(Feature Store)是使用最广泛的开源特征存储，同时支持离线(批量训练)和在线(实时服务)两条路径。本文将介绍用 Feast 构建特征流水线的完整过程。

为什么需要 Feature Store

Training-Serving Skew 问题

# 训练时(离线)
features = pd.read_sql("""
    SELECT user_id,
           AVG(purchase_amount) as avg_purchase,
           COUNT(*) as purchase_count
    FROM transactions
    WHERE timestamp < '2026-01-01'
    GROUP BY user_id
""", conn)

# 服务时(在线) - 用不同的逻辑计算就会产生 Skew！
features = redis_client.get(f"user:{user_id}:features")

训练和服务用不同的代码计算同一个特征，就会产生细微的差异，模型性能也会偏离离线实验的结果。Feature Store 从单一的特征定义出发，为离线和在线两侧都提供一致的值。

Feature Store 的核心功能

功能说明
特征注册表管理特征的元数据、schema、所有者
离线存储面向批量训练的大量特征查询(Point-in-Time Join）
在线存储面向实时服务的低延迟特征查询
特征服务通过 gRPC/HTTP API 提供特征服务
Point-in-Time Join按时间基准精确关联特征值

安装 Feast 并初始化项目

安装

# 基础安装
pip install feast

# 使用 PostgreSQL 在线存储时
pip install feast[postgres]

# 使用 Redis 在线存储时
pip install feast[redis]

# 安装全部依赖
pip install feast[postgres,redis,aws,gcp]

项目初始化

# 创建项目
feast init my_feature_store
cd my_feature_store

# 目录结构
# my_feature_store/
# ├── feature_repo/
# │   ├── feature_store.yaml    # Feast 配置
# │   ├── example_repo.py       # 特征定义示例
# │   └── data/                 # 示例数据
# └── README.md

feature_store.yaml 配置

project: my_feature_store
registry: data/registry.db
provider: local

online_store:
  type: sqlite
  path: data/online_store.db

offline_store:
  type: file

entity_key_serialization_version: 2

生产环境中按如下方式修改：

project: my_feature_store
registry:
  registry_type: sql
  path: postgresql://user:pass@host:5432/feast_registry

provider: local

online_store:
  type: redis
  connection_string: redis://localhost:6379

offline_store:
  type: file # 或 bigquery、redshift、snowflake

特征定义

数据源与实体定义

# feature_repo/features.py
from datetime import timedelta
from feast import Entity, FeatureView, Field, FileSource, PushSource
from feast.types import Float32, Int64, String

# 数据源定义
user_transactions_source = FileSource(
    path="data/user_transactions.parquet",
    timestamp_field="event_timestamp",
    created_timestamp_column="created_timestamp",
)

# 实体定义(作为特征基准的键)
user = Entity(
    name="user_id",
    join_keys=["user_id"],
    description="用户唯一 ID",
)

Feature View 定义

# 离线 + 在线特征视图
user_transaction_features = FeatureView(
    name="user_transaction_features",
    entities=[user],
    ttl=timedelta(days=7),  # 在线存储中 7 天后过期
    schema=[
        Field(name="total_purchases", dtype=Int64, description="总购买次数"),
        Field(name="avg_purchase_amount", dtype=Float32, description="平均购买金额"),
        Field(name="last_purchase_amount", dtype=Float32, description="最近一次购买金额"),
        Field(name="purchase_frequency", dtype=Float32, description="购买频率(笔/天)"),
        Field(name="user_segment", dtype=String, description="用户分层"),
    ],
    online=True,
    source=user_transactions_source,
    tags={"team": "ml-platform", "version": "v1"},
)

On-Demand Feature View(实时转换)

from feast import on_demand_feature_view, RequestSource

# 请求时动态计算的特征
input_request = RequestSource(
    name="purchase_request",
    schema=[
        Field(name="current_amount", dtype=Float32),
    ],
)

@on_demand_feature_view(
    sources=[user_transaction_features, input_request],
    schema=[
        Field(name="amount_vs_avg_ratio", dtype=Float32),
        Field(name="is_high_value", dtype=Int64),
    ],
)
def purchase_analysis(inputs: dict) -> dict:
    """计算本次购买金额与平均购买金额的比率"""
    import pandas as pd
    df = pd.DataFrame(inputs)
    df["amount_vs_avg_ratio"] = df["current_amount"] / (df["avg_purchase_amount"] + 1e-6)
    df["is_high_value"] = (df["amount_vs_avg_ratio"] > 2.0).astype(int)
    return df[["amount_vs_avg_ratio", "is_high_value"]]

生成示例数据

# scripts/generate_data.py
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

np.random.seed(42)
n_users = 1000
n_records = 5000

user_ids = [f"user_{i:04d}" for i in range(n_users)]
records = []

for _ in range(n_records):
    user_id = np.random.choice(user_ids)
    ts = datetime(2026, 1, 1) + timedelta(
        days=np.random.randint(0, 60),
        hours=np.random.randint(0, 24),
    )
    records.append({
        "user_id": user_id,
        "total_purchases": np.random.randint(1, 100),
        "avg_purchase_amount": round(np.random.uniform(10, 500), 2),
        "last_purchase_amount": round(np.random.uniform(5, 1000), 2),
        "purchase_frequency": round(np.random.uniform(0.1, 5.0), 3),
        "user_segment": np.random.choice(["bronze", "silver", "gold", "platinum"]),
        "event_timestamp": ts,
        "created_timestamp": ts,
    })

df = pd.DataFrame(records)
df.to_parquet("feature_repo/data/user_transactions.parquet", index=False)
print(f"Generated {len(df)} records for {n_users} users")

python scripts/generate_data.py
# Generated 5000 records for 1000 users

Feast 工作流

1. Apply — 注册特征定义

cd feature_repo
feast apply

Created entity user_id
Created feature view user_transaction_features
Created on demand feature view purchase_analysis

Deploying infrastructure for my_feature_store...

2. Materialize — 从离线存储同步到在线存储

# 把指定时间段的数据加载到在线存储
feast materialize 2026-01-01T00:00:00 2026-03-01T00:00:00

# 增量加载(从上次 materialize 到当前)
feast materialize-incremental $(date -u +"%Y-%m-%dT%H:%M:%S")

Materializing 1 feature views from 2026-01-01 to 2026-03-01
user_transaction_features:
100%|████████████████████████| 1000/1000 [00:03<00:00, 312.45it/s]

3. 离线特征查询(用于训练)

from feast import FeatureStore
import pandas as pd

store = FeatureStore(repo_path="feature_repo")

# 用于生成训练数据的实体 DataFrame
entity_df = pd.DataFrame({
    "user_id": ["user_0001", "user_0042", "user_0100", "user_0500"],
    "event_timestamp": pd.to_datetime([
        "2026-02-01", "2026-02-15", "2026-01-20", "2026-02-28"
    ]),
})

# 通过 Point-in-Time Join 查询特征
training_df = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "user_transaction_features:total_purchases",
        "user_transaction_features:avg_purchase_amount",
        "user_transaction_features:last_purchase_amount",
        "user_transaction_features:purchase_frequency",
        "user_transaction_features:user_segment",
    ],
).to_df()

print(training_df.head())

    user_id  event_timestamp  total_purchases  avg_purchase_amount  ...
0  user_0001  2026-02-01            45           234.56             ...
1  user_0042  2026-02-15            12            89.30             ...
2  user_0100  2026-01-20            78           456.78             ...
3  user_0500  2026-02-28            33           167.42             ...

Point-in-Time Join 是这里的核心。它会取出每个实体在其 event_timestamp 时点上最新的特征值。这样就能在不发生数据泄漏(data leakage)的前提下，构建出准确的训练数据。

4. 在线特征查询(用于服务)

# 在实时服务中查询特征
online_features = store.get_online_features(
    features=[
        "user_transaction_features:total_purchases",
        "user_transaction_features:avg_purchase_amount",
        "user_transaction_features:user_segment",
        "purchase_analysis:amount_vs_avg_ratio",
        "purchase_analysis:is_high_value",
    ],
    entity_rows=[
        {"user_id": "user_0001", "current_amount": 750.0},
        {"user_id": "user_0042", "current_amount": 50.0},
    ],
).to_dict()

print(online_features)

{
    "user_id": ["user_0001", "user_0042"],
    "total_purchases": [45, 12],
    "avg_purchase_amount": [234.56, 89.30],
    "user_segment": ["gold", "silver"],
    "amount_vs_avg_ratio": [3.199, 0.560],
    "is_high_value": [1, 0],
}

用 Feature Service 管理特征分组

from feast import FeatureService

# 推荐模型所需的特征组合
recommendation_service = FeatureService(
    name="recommendation_features",
    features=[
        user_transaction_features[["total_purchases", "avg_purchase_amount", "user_segment"]],
        purchase_analysis,
    ],
    tags={"model": "recommendation-v2"},
)

# 欺诈检测模型所需的特征组合
fraud_detection_service = FeatureService(
    name="fraud_detection_features",
    features=[
        user_transaction_features,
        purchase_analysis,
    ],
    tags={"model": "fraud-detection-v1"},
)

# 通过 Feature Service 查询
features = store.get_online_features(
    features=store.get_feature_service("recommendation_features"),
    entity_rows=[{"user_id": "user_0001", "current_amount": 750.0}],
).to_dict()

用 Push Source 实时更新特征

from feast import PushSource

# Push 数据源定义
user_realtime_source = PushSource(
    name="user_realtime_push",
    batch_source=user_transactions_source,
)

# 实时事件发生时更新特征
store.push(
    push_source_name="user_realtime_push",
    df=pd.DataFrame({
        "user_id": ["user_0001"],
        "total_purchases": [46],
        "avg_purchase_amount": [240.12],
        "last_purchase_amount": [750.0],
        "purchase_frequency": [2.1],
        "user_segment": ["gold"],
        "event_timestamp": [pd.Timestamp.now()],
        "created_timestamp": [pd.Timestamp.now()],
    }),
)

部署 Feature Server

# 启动本地 Feature Server
feast serve -h 0.0.0.0 -p 6566

# 通过 HTTP API 查询特征
curl -X POST http://localhost:6566/get-online-features \
  -H "Content-Type: application/json" \
  -d '{
    "features": [
      "user_transaction_features:total_purchases",
      "user_transaction_features:avg_purchase_amount"
    ],
    "entities": {
      "user_id": ["user_0001", "user_0042"]
    }
  }'

用 Docker 部署 Feature Server

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install feast[redis]

COPY feature_repo/ feature_repo/
WORKDIR /app/feature_repo

# 应用 Registry 并启动服务
CMD feast apply && feast serve -h 0.0.0.0 -p 6566

# docker-compose.yml
services:
  feast-server:
    build: .
    ports:
      - '6566:6566'
    depends_on:
      - redis
    environment:
      - REDIS_URL=redis://redis:6379

  redis:
    image: redis:7-alpine
    ports:
      - '6379:6379'

与 Airflow 集成(自动 Materialize)

# dags/feast_materialize.py
from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

default_args = {
    "owner": "ml-platform",
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
}

with DAG(
    dag_id="feast_materialize",
    default_args=default_args,
    schedule_interval="0 */6 * * *",  # 每 6 小时
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:

    materialize = BashOperator(
        task_id="materialize_incremental",
        bash_command=(
            "cd /opt/feature_repo && "
            "feast materialize-incremental $(date -u +'%Y-%m-%dT%H:%M:%S')"
        ),
    )

结语

梳理一下用 Feast 构建特征流水线的核心要点：

  • 一致的特征定义：训练和服务使用同一份特征定义，防止 Training-Serving Skew
  • Point-in-Time Join：按时间基准精确关联特征，防止数据泄漏
  • 离线/在线双存储：批量训练用离线存储，实时服务用在线存储
  • Feature Service：按模型分组管理特征，提升复用性
  • Push Source：支持基于实时事件的特征更新

当 ML 模型只有一两个的时候，Feature Store 可能会显得有点小题大做，但随着模型增多、团队变大，它会成为必不可少的基础设施。尤其是当多个模型共享同一批特征时，它的价值才会被最大化。

测验

Q1: Training-Serving Skew 是什么？ 训练时使用的特征和服务时使用的特征不一致，导致模型性能下降的现象。特征计算逻辑的不一致、数据源的差异、 时间基准的不一致等都是原因。

Q2: Point-in-Time Join 的作用是什么？以每个实体的事件时点(event_timestamp)为基准， 关联该时点之前最新的特征值。这样可以防止未来数据被用于训练所导致的数据泄漏(data leakage)。

Q3: Feast 中离线存储和在线存储的区别是什么？ 离线存储保存大量的历史特征，用于批量训练(文件、BigQuery 等)；在线存储只保存最新的特征值， 用于低延迟的实时服务(Redis、DynamoDB 等)。

Q4: feast materialize 命令的作用是什么？ 把离线存储中的特征数据同步(加载)到在线存储的操作。它会把指定时间范围内的最新特征值存入在线存储， 使实时查询成为可能。

Q5: On-Demand Feature View 和普通 Feature View 的区别是什么？ 普通 Feature View 保存的是预先计算好的特征，而 On-Demand Feature View 在请求时点动态计算特征。 它用于结合请求参数和已有特征的实时转换。

Q6: Feature Service 的优点是什么？ 可以按模型对所需特征进行逻辑分组管理。既能清楚地追踪哪个模型在用哪些特征，也能在查询特征时提供 一致的接口。

Q7: TTL(Time To Live)设置的含义是什么？ 指定在线存储中特征值的有效期。超过 TTL 的特征在查询时会返回 null，从而防止过期(stale)的特征值 被用于服务。

Q8: 什么场景下会用到 Push Source？ 当实时事件(支付、点击等)发生时，需要立即更新在线存储中的特征的场景。它能在批量 materialize 的 定期更新之间，维持特征的最新状态。

현재 단락 (1/292)

把 ML 模型部署到生产环境时，最常见的问题之一就是**Training-Serving Skew**(训练-服务偏差)。训练时使用的特征和服务时使用的特征不一致，导致模型性能下降的现象。**Feat...

작성 글자: 0자원문 글자: 9,739자작성 단락: 0/292