필사 모드: Feature Store 构建完全指南：Feast 架构、在线/离线服务与 ML 流水线集成

# 安装 Feast（含 Redis、PostgreSQL 支持）
pip install 'feast[redis,postgres]'

# 初始化项目
feast init feature_repo
cd feature_repo

# 目录结构
# feature_repo/
#   feature_store.yaml    -- 项目配置
#   definitions.py        -- Entity、Feature View 定义
#   data/                 -- 示例数据

project: my_ml_platform
provider: gcp
registry:
  registry_type: sql
  path: postgresql://feast:feast@db-host:5432/feast_registry
  cache_ttl_seconds: 60
online_store:
  type: redis
  connection_string: redis-host:6379,password=secret
offline_store:
  type: bigquery
  dataset: feast_offline
entity_key_serialization_version: 2

from datetime import timedelta
from feast import Entity, FeatureView, Field, FileSource, BigQuerySource
from feast.types import Float32, Float64, Int64, String

# Entity 定义
customer = Entity(
    name="customer_id",
    description="顾客的唯一标识符",
)

driver = Entity(
    name="driver_id",
    description="司机的唯一标识符",
)

# BigQuery 数据源定义
customer_stats_source = BigQuerySource(
    name="customer_stats_source",
    table="my_project.feast_dataset.customer_stats",
    timestamp_field="event_timestamp",
    created_timestamp_column="created_timestamp",
)

# Feature View 定义
customer_stats_fv = FeatureView(
    name="customer_stats",
    entities=[customer],
    ttl=timedelta(days=3),
    schema=[
        Field(name="total_orders", dtype=Int64),
        Field(name="avg_order_amount", dtype=Float64),
        Field(name="lifetime_value", dtype=Float64),
        Field(name="preferred_category", dtype=String),
        Field(name="churn_risk_score", dtype=Float32),
    ],
    source=customer_stats_source,
    online=True,
    tags={
        "team": "growth",
        "version": "v2",
    },
)

driver_stats_source = BigQuerySource(
    name="driver_stats_source",
    table="my_project.feast_dataset.driver_stats",
    timestamp_field="event_timestamp",
)

driver_stats_fv = FeatureView(
    name="driver_stats",
    entities=[driver],
    ttl=timedelta(hours=6),
    schema=[
        Field(name="avg_rating", dtype=Float64),
        Field(name="total_trips", dtype=Int64),
        Field(name="acceptance_rate", dtype=Float64),
        Field(name="avg_delivery_time_min", dtype=Float32),
    ],
    source=driver_stats_source,
    online=True,
)

from feast import FeatureService

# 用于流失预测模型的 Feature Service
churn_prediction_svc = FeatureService(
    name="churn_prediction_service",
    features=[
        customer_stats_fv[["total_orders", "avg_order_amount", "lifetime_value", "churn_risk_score"]],
    ],
    tags={
        "model": "churn_prediction_v3",
        "owner": "growth-team",
    },
)

# 用于司机匹配模型的 Feature Service
driver_matching_svc = FeatureService(
    name="driver_matching_service",
    features=[
        driver_stats_fv[["avg_rating", "acceptance_rate", "avg_delivery_time_min"]],
        customer_stats_fv[["preferred_category"]],
    ],
)

from feast import FeatureStore
import pandas as pd

store = FeatureStore(repo_path="feature_repo/")

# 训练用实体 DataFrame（含标签）
entity_df = pd.DataFrame({
    "customer_id": [1001, 1002, 1003, 1001, 1002],
    "event_timestamp": pd.to_datetime([
        "2026-01-15 10:00:00",
        "2026-01-15 11:00:00",
        "2026-01-16 09:00:00",
        "2026-02-01 10:00:00",
        "2026-02-01 11:00:00",
    ]),
    "churned": [0, 1, 0, 1, 0],  # 标签
})

# 通过 Point-in-Time Join 生成训练数据
training_df = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "customer_stats:total_orders",
        "customer_stats:avg_order_amount",
        "customer_stats:lifetime_value",
        "customer_stats:churn_risk_score",
    ],
).to_df()

print(training_df.head())
# customer_id | event_timestamp     | churned | total_orders | avg_order_amount | ...
# 1001        | 2026-01-15 10:00:00 | 0       | 42           | 35.50            | ...

# 在线特征查询
online_features = store.get_online_features(
    features=[
        "customer_stats:total_orders",
        "customer_stats:avg_order_amount",
        "customer_stats:churn_risk_score",
    ],
    entity_rows=[
        {"customer_id": 1001},
        {"customer_id": 1002},
    ],
).to_dict()

print(online_features)
# 输出示例：
# {
#     "customer_id": [1001, 1002],
#     "total_orders": [45, 12],
#     "avg_order_amount": [35.50, 28.00],
#     "churn_risk_score": [0.15, 0.82],
# }

# 全量 Feature View Materialization
feast materialize 2026-01-01T00:00:00 2026-03-10T00:00:00

# 增量 Materialization（仅同步上次运行之后的变化）
feast materialize-incremental 2026-03-10T00:00:00

from airflow import DAG
from airflow.decorators import task
from datetime import datetime, timedelta

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

with DAG(
    dag_id="feast_materialization_pipeline",
    default_args=default_args,
    schedule_interval="0 */4 * * *",  # 每 4 小时运行一次
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:

    @task()
    def validate_source_data():
        """验证源数据质量"""
        from great_expectations import get_context
        context = get_context()
        result = context.run_checkpoint(checkpoint_name="feature_source_check")
        if not result.success:
            raise ValueError("源数据质量验证失败")
        return True

    @task()
    def materialize_features():
        """从离线存储向在线存储执行 Materialization"""
        from feast import RepoConfig, FeatureStore
        from feast.infra.online_stores.redis import RedisOnlineStoreConfig
        from feast.repo_config import RegistryConfig

        repo_config = RepoConfig(
            project="my_ml_platform",
            provider="gcp",
            registry=RegistryConfig(
                registry_type="sql",
                path="postgresql://feast:feast@db-host:5432/feast_registry",
            ),
            online_store=RedisOnlineStoreConfig(
                connection_string="redis-host:6379",
            ),
        )
        store = FeatureStore(config=repo_config)
        store.materialize_incremental(end_date=datetime.utcnow())
        return True

    @task()
    def validate_online_store():
        """验证在线存储的特征值"""
        from feast import FeatureStore
        store = FeatureStore(repo_path="feature_repo/")

        # 使用示例实体查询在线特征
        result = store.get_online_features(
            features=["customer_stats:total_orders"],
            entity_rows=[{"customer_id": 1001}],
        ).to_dict()

        if result["total_orders"][0] is None:
            raise ValueError("特征未加载到在线存储")
        return True

    @task()
    def notify_completion():
        """发送 Slack 通知"""
        import requests
        requests.post(
            "https://hooks.slack.com/services/YOUR/WEBHOOK/URL",
            json={"text": "Feast Materialization 已完成"},
        )

    validate_source_data() >> materialize_features() >> validate_online_store() >> notify_completion()

from kfp import dsl
from kfp.dsl import component, Output, Dataset

@component(
    base_image="python:3.10",
    packages_to_install=["feast[redis,postgres]>=0.40.0"],
)
def feast_materialize_op(
    project_name: str,
    registry_path: str,
    redis_connection: str,
):
    from feast import RepoConfig, FeatureStore
    from feast.infra.online_stores.redis import RedisOnlineStoreConfig
    from feast.repo_config import RegistryConfig
    from datetime import datetime

    config = RepoConfig(
        project=project_name,
        provider="gcp",
        registry=RegistryConfig(
            registry_type="sql",
            path=registry_path,
        ),
        online_store=RedisOnlineStoreConfig(
            connection_string=redis_connection,
        ),
    )
    store = FeatureStore(config=config)
    store.materialize_incremental(end_date=datetime.utcnow())

@component(
    base_image="python:3.10",
    packages_to_install=["feast[redis,postgres]>=0.40.0", "scikit-learn"],
)
def train_model_op(
    project_name: str,
    model_output: Output[Dataset],
):
    from feast import FeatureStore
    import pandas as pd
    from sklearn.ensemble import GradientBoostingClassifier
    import pickle

    store = FeatureStore(repo_path="feature_repo/")
    entity_df = pd.read_parquet("gs://my-bucket/training_entities.parquet")

    training_df = store.get_historical_features(
        entity_df=entity_df,
        features=[
            "customer_stats:total_orders",
            "customer_stats:avg_order_amount",
            "customer_stats:churn_risk_score",
        ],
    ).to_df()

    X = training_df.drop(columns=["customer_id", "event_timestamp", "churned"])
    y = training_df["churned"]

    model = GradientBoostingClassifier(n_estimators=200)
    model.fit(X, y)

    with open(model_output.path, "wb") as f:
        pickle.dump(model, f)

@dsl.pipeline(name="feast-ml-training-pipeline")
def feast_training_pipeline():
    materialize_task = feast_materialize_op(
        project_name="my_ml_platform",
        registry_path="postgresql://feast:feast@db-host:5432/feast_registry",
        redis_connection="redis-host:6379",
    )
    train_task = train_model_op(
        project_name="my_ml_platform",
    )
    train_task.after(materialize_task)