Feast 特征仓库实战指南从实体定义到向量检索与 RAG 的完整工作流【免费下载链接】feastThe Open Source Feature Store for AI/ML项目地址: https://gitcode.com/GitHub_Trending/fe/feastFeastFeature Store是一个开源特征存储系统为 AI/ML 场景提供特征定义、注册、物化与在线/离线检索的一站式能力。本篇指南以 skills/SKILL.md 为核心骨架系统讲解在 Feast 项目中定义实体Entity、数据源DataSource、特征视图FeatureView与特征服务FeatureService通过feast apply注册定义再以 CLI 或 Python SDK 完成物化、在线低延迟特征检索、基于时间点的历史训练数据生成以及基于向量索引的相似度检索与 RAG 问答流水线。读完本文你将能够从零搭建一个可运行的 Feast 特征仓库并理解每个环节对应的源码实现位置便于进一步深入项目内部。Quick Start五分钟跑通一个 Feast 项目一个最小的 Feast 项目只需要三样东西一份feature_store.yaml配置文件若干 Python 文件用于定义实体、数据源、特征视图和特征服务执行feast apply将定义注册进 Registry。feast init my_project cd my_project feast applyfeast init会基于 sdk/python/feast/templates 下的模板生成一个可直接运行的示例仓库包含feature_store.yaml与示例特征定义文件。feast apply会递归扫描仓库内所有.py文件发现其中的实体、特征视图、特征服务等定义对象统一注册进 Registry默认是本地文件data/registry.db由 sdk/python/feast/feature_store.py 的apply方法执行。之后你就能立即使用 Python SDK 检索特征了。核心概念与定义方式Entity特征的关联主体实体Entity是一组语义相关特征的集合例如一位客户或一名司机。实体通过 join key 与特征进行关联查询。from feast import Entity from feast.value_type import ValueType driver Entity( namedriver_id, descriptionDriver identifier, value_typeValueType.INT64, )从 sdk/python/feast/entity.py 的Entity.__init__源码可以看到其参数语义name是全局唯一实体名join_keys用于查询时定位实体当前仅支持单元素列表源码中会校验len(join_keys) 1并抛出ValueError多 join key 支持仍在规划中不指定时默认取name本身value_type如果不填源码会发出DeprecationWarning提示后续版本将强制要求并从底层数据源的 schema 中推断此外还支持description、tags、owner等元数据字段。数据源原始特征数据从哪来数据源DataSource描述原始特征数据的存放位置Feast 按批/流两类区分from feast import FileSource, BigQuerySource, KafkaSource, PushSource, RequestSource from feast.data_format import ParquetFormat # 批数据源本地文件 driver_stats_source FileSource( namedriver_stats_source, pathdata/driver_stats.parquet, timestamp_fieldevent_timestamp, created_timestamp_columncreated, ) # 请求数据源供 OnDemandFeatureView 使用 input_request RequestSource( namevals_to_add, schema[Field(nameval_to_add, dtypeFloat64)], )批数据源FileSourceParquet/CSV 本地文件、BigQuerySourceGCP、SnowflakeSource、RedshiftSource、PostgreSQLSource、SparkSource、TrinoSource、AthenaSource、ClickhouseSource等声明位置可在 sdk/python/feast/data_source.py 与 sdk/python/feast/infra/offline_stores/contrib 下逐一找到流数据源KafkaSource需提供kafka_bootstrap_servers、topic、message_format并用batch_source指定回填用批数据源、KinesisSourceregion、stream_name、record_formatPushSource供 SDK 侧手动推送数据如实时特征使用RequestSource不落盘仅承载请求时传入的参数是构造 OnDemandFeatureView 的基础输入之一。FeatureView特征的核心定义单元特征视图FeatureView把数据源中的字段映射到实体上同时声明 schema、TTL 以及在线/离线可用性from feast import FeatureView, Field from feast.types import Float32, Int64, String from datetime import timedelta driver_hourly_stats FeatureView( namedriver_hourly_stats, entities[driver], ttltimedelta(days365), schema[ Field(nameconv_rate, dtypeFloat32), Field(nameacc_rate, dtypeFloat32), Field(nameavg_daily_trips, dtypeInt64), ], onlineTrue, sourcedriver_stats_source, )关键参数source必需批或流数据源、entities关联实体列表、schema特征字段列表不声明时 Feast 会从数据源自动推断、ttl特征有效期默认timedelta(0)用于控制历史特征保留与在线存储清理、online默认True是否可用于在线检索、offline默认False是否可用于离线检索以及mode变换模式可选python、pandas、sql、spark、ray、substrait实现类可参见 sdk/python/feast/feature_view.py 与 sdk/python/feast/batch_feature_view.py。Field 与类型系统Field 是特征字段的载体除name、dtype外还支持vector_index设为True时为该字段建立向量索引以支持相似度检索、vector_length向量维度vector_indexTrue时必须给出、vector_search_metric相似度度量等向量相关属性源码中均有对应构造参数。Feast 的类型体系位于 sdk/python/feast/types.py分为三层标量类型feast.typesFloat32、Float64、Int32、Int64、String、Bool、Bytes、UnixTimestamp、ZonedTimestamp集合类型Array(T)其中T为标量类型例如Array(Float32)常用于表示 embedding 向量遗留 ValueType 枚举feast.value_typeSTRING、INT32、INT64、FLOAT、DOUBLE、BOOL、BYTES、UNIX_TIMESTAMP及_LIST/_SET变体主要用于Entity.value_type等旧接口。Python 原生类型到 Feast 类型的映射约定为int→INT64、str→STRING、float→DOUBLE、bytes→BYTES、bool→BOOL、datetime→UNIX_TIMESTAMP。OnDemandFeatureView请求时实时变换按需特征视图ODFV在请求到达时基于其他特征视图的输出和/或请求参数动态计算特征适合在线实时变换、模型上线前的特征组合等场景。装饰器定义在 sdk/python/feast/on_demand_feature_view.py默认modepandas。from feast import on_demand_feature_view import pandas as pd on_demand_feature_view( sources[driver_hourly_stats, input_request], schema[Field(nameconv_rate_plus_val, dtypeFloat64)], modepandas, ) def transformed_conv_rate(inputs: pd.DataFrame) - pd.DataFrame: df pd.DataFrame() df[conv_rate_plus_val] inputs[conv_rate] inputs[val_to_add] return dfODFV 支持四种模式Pandas 模式默认函数接收pd.DataFrame并返回pd.DataFrame适合向量化运算Python 模式modepython函数接收字典、返回字典逐行逻辑更直观例如把连续值分箱成类别Python 单例模式modepython, singletonTrue单行输入时输入输出均为单值字典无需维护列表写回在线存储write_to_online_storeTrue在写入时预计算并直接落入在线存储而非读取时实时计算适用于计算代价高、希望降低检索延迟的场景。此外 ODFV 还支持aggregations参数预变换聚合以及基于Aggregation(column..., function...)的聚合式 ODFV。注意feast apply默认会用随机输入校验 ODFV 逻辑若自定义逻辑复杂导致校验失败可加--skip-feature-view-validation跳过。StreamFeatureView流式特征与窗口聚合流特征视图SFV继承自 FeatureView专为流数据源Kafka、Kinesis、PushSource设计可搭配Aggregation做时间窗口聚合from feast import StreamFeatureView, Field from feast.types import Int64 from feast.aggregation import Aggregation from datetime import timedelta driver_stream StreamFeatureView( namedriver_trips_stream, entities[driver], sourcekafka_source, schema[Field(nametrips, dtypeInt64)], ttltimedelta(hours2), aggregations[ Aggregation(columntrips, functioncount, time_windowtimedelta(hours1)), ], timestamp_fieldevent_timestamp, )Aggregation的function支持sum、max、min、count、meantime_window指定聚合窗口slide_interval用于滑动窗口的滑动步长。SFV 的udf参数还可传入变换函数mode支持python、pandas、spark、spark_sql。FeatureService把特征打包成可复用的服务特征服务把来自一个或多个特征视图的特征打包成组便于模型训练与在线推理时统一引用避免在业务代码里逐个写特征引用from feast import FeatureService driver_fs FeatureService( namedriver_ranking, features[driver_hourly_stats, transformed_conv_rate], )也支持用投影语法只选取部分字段driver_hourly_stats[[conv_rate, acc_rate]]生成的投影对象由 sdk/python/feast/feature_view_projection.py 的FeatureViewProjection承载。特征检索在线低延迟与历史训练数据通过 FeatureStore 统一入口执行检索。FeatureStore有三种构造方式指定repo_path自动寻找feature_store.yaml、直接传入RepoConfig配置对象、或通过fs_yaml_file指向显式 YAML 路径。在线检索低延迟在线特征必须先用feast materialize物化到在线存储之后才能以毫秒级延迟按实体键查询最新值from feast import FeatureStore store FeatureStore(repo_path.) features store.get_online_features( features[ driver_hourly_stats:conv_rate, driver_hourly_stats:acc_rate, ], entity_rows[{driver_id: 1001}, {driver_id: 1002}], ).to_dict()特征引用采用特征视图名:特征名格式例如driver_hourly_stats:conv_rate。get_online_features还支持full_feature_namesTrue让返回列名带上视图前缀如driver_hourly_stats__conv_rate避免多视图同名字段冲突。从源码可见该方法内部会委托给当前 provider 的get_online_features实现并支持可选的 MLflow 自动日志配置mlflow.auto_log后记录检索耗时、实体数与特征服务名详见 sdk/python/feast/feature_store.py。首次调用时若 Registry 尚未缓存会先下载完整 Registry耗时取决于远程 RegistryGCS/S3网络情况Registry 默认带 600 秒 TTL 缓存可通过refresh_registry()主动刷新。在线检索也可以直接传 FeatureService 作为features参数一次拿回服务内全部特征。历史检索训练数据 时间点正确连接历史检索返回的是训练数据集其核心价值是时间点正确连接point-in-time joinFeast 会根据实体事件时间戳只关联该时刻之前已生效的特征杜绝标签泄漏import pandas as pd from datetime import datetime entity_df pd.DataFrame({ driver_id: [1001, 1002], event_timestamp: [datetime(2023, 1, 1), datetime(2023, 1, 2)], }) training_df store.get_historical_features( entity_dfentity_df, features[driver_hourly_stats:conv_rate, driver_hourly_stats:acc_rate], ).to_df()返回的RetrievalJob提供.to_df()pandas DataFrame、.to_arrow()PyArrow Table、.to_sql_string()对 SQL 型离线存储返回对应查询 SQL三种取数方式。同样地features参数也可以替换为 FeatureServicetraining_df store.get_historical_features( entity_dfentity_df, featuresdriver_fs, ).to_df()推送与直写对于 PushSource / StreamFeatureView 场景可用store.push(push_source_name..., df...)将新数据推入在线与离线存储需要直接控制写路径时可用write_to_online_store(feature_view_name..., df...)与write_to_offline_store(feature_view_name..., df...)分别直写。物化把特征从离线同步到在线物化Materialization把离线存储如 Parquet、BigQuery、Snowflake中的特征值按实体与时间范围批量加载到在线存储SQLite、Redis、DynamoDB 等是离线可训练、在线可服务的关键一环# 全量物化覆盖指定时间范围 feast materialize 2023-01-01T00:00:00 2023-12-31T23:59:59 # 增量物化从上一次物化位置续传至今 feast materialize-incremental $(date -u %Y-%m-%dT%H:%M:%S)Python API 等价写法from datetime import datetime store.materialize(start_datedatetime(2023, 1, 1), end_datedatetime(2023, 12, 31)) store.materialize_incremental(end_datedatetime.utcnow())物化引擎由batch_engine决定默认local本地 Python 进程还支持snowflake.engine、spark.engine、lambdaAWS Lambda、k8sKubernetes Job、ray.engine等远程/分布式引擎。CLI 命令速查feastCLI 是管理特征仓库的主要入口命令实现位于 sdk/python/feast/cli命令用途feast init [DIR]创建新的特征仓库feast apply注册/更新特征定义到 Registryfeast plan预览将要发生的变更而不实际应用feast materialize START END将时间范围内的特征物化到在线存储feast materialize-incremental END增量物化feast entities list列出已注册实体feast feature-views list列出特征视图feast feature-services list列出特征服务feast on-demand-feature-views list列出按需特征视图feast teardown清理基础设施资源feast version显示 SDK 版本通用选项--chdir/-c指定在其他目录执行--feature-store-yaml/-f覆盖默认feature_store.yaml的路径。向量检索与 RAG 流水线Feast 原生支持把 embedding 作为特征字段建索引进而支撑语义检索与检索增强生成RAG。定义带向量索引的特征视图from feast.types import Array, Float32 wiki_passages FeatureView( namewiki_passages, entities[passage_entity], schema[ Field(namepassage_text, dtypeString), Field( nameembedding, dtypeArray(Float32), vector_indexTrue, vector_length384, vector_search_metricCOSINE, ), ], sourcepassages_source, onlineTrue, )要求底层在线存储支持向量检索例如 Milvustype: milvus、Qdranttype: qdrant或带 pgvector 的 PostgreSQL。相似度度量支持三种COSINE— 余弦相似度默认值适合归一化 embeddingL2— 欧氏距离INNER_PRODUCT— 内积。检索相似文档# v1 API直接传 query 向量 results store.retrieve_online_documents( featurewiki_passages:embedding, queryquery_embedding, top_k5, ) # v2 API支持文本、向量、图像三类查询 results store.retrieve_online_documents_v2( feature_view_namewiki_passages, query_stringWhat is machine learning?, top_k5, )两个方法的实现位于 sdk/python/feast/feature_store.pyretrieve_online_documents与同文件retrieve_online_documents_v2底层向量检索逻辑可深入 sdk/python/feast/vector_store.py 查看。用 FeastRAGRetriever 搭建问答流水线FeastRAGRetriever 将 Feast 与 HuggingFace 生态打通先用 DPR 之类的 question encoder 把问题编码为向量到 Feast 在线存储中检索最相似的文档片段再交给生成模型如 BART产出答案。前置条件一个含vector_indexTrueembedding 字段的 FeatureView、特征已物化到在线存储、已安装 HuggingFacetransformers。from feast.rag_retriever import FeastRAGRetriever from transformers import AutoTokenizer, AutoModel, AutoModelForSeq2SeqLM question_tokenizer AutoTokenizer.from_pretrained(facebook/dpr-question_encoder-single-nq-base) question_encoder AutoModel.from_pretrained(facebook/dpr-question_encoder-single-nq-base) generator_tokenizer AutoTokenizer.from_pretrained(facebook/bart-large) generator_model AutoModelForSeq2SeqLM.from_pretrained(facebook/bart-large) retriever FeastRAGRetriever( question_encoder_tokenizerquestion_tokenizer, question_encoderquestion_encoder, generator_tokenizergenerator_tokenizer, generator_modelgenerator_model, feast_repo_pathpath/to/feature_repo, feature_viewwiki_passages, features[passage_text, embedding], search_typevector, # text、vector 或 hybrid id_fieldpassage_id, text_fieldpassage_text, )检索文档片段doc_embeddings, doc_ids, doc_dicts retriever.retrieve( question_input_idsquestion_tokenizer(What is ML?, return_tensorspt)[input_ids], n_docs5, )端到端生成答案answer retriever.generate_answer( queryWhat is machine learning?, top_k5, max_new_tokens200, ) print(answer)若只需低层向量检索能力也可直接使用FeastVectorStore位于 sdk/python/feast/vector_store.py其query()方法支持query_vector、query_string、query_image_bytes三种查询形态。feature_store.yaml 配置详解feature_store.yaml是特征仓库的配置文件对应RepoConfig定义于 sdk/python/feast/repo_config.py。最小本地配置project: my_project registry: data/registry.db provider: local online_store: type: sqlite path: data/online_store.dbRepoConfig 核心字段字段别名类型默认值说明project-str必填项目命名空间字母数字 下划线project_description-strNone项目描述provider-strlocallocal、gcp或awsregistryregistry_configstr/dict必填Registry 路径或配置对象online_storeonline_configstr/dictsqlite在线存储类型或配置offline_storeoffline_configstr/dictdask离线存储类型或配置batch_enginebatch_engine_configstr/dictlocal批物化引擎auth-dictno_auth认证配置feature_server-dictNone特征服务器配置entity_key_serialization_version-int3实体键序列化版本coerce_tz_aware-boolTrue将时间戳统一为带时区格式materializationmaterialization_configdict默认物化选项openlineageopenlineage_configdictNoneOpenLineage 血缘配置Registry 配置registry支持四类registry_typefile默认本地路径或 GCS/S3 URI、sql数据库连接 URL、snowflake.registry、remote指向远程 Registry 服务。常用字段包括cache_ttl_seconds默认 6000 表示永不过期、cache_modesync或thread、s3_additional_kwargs。# 文件型简写 registry: data/registry.db # 文件型完整写法 registry: registry_type: file path: data/registry.db cache_ttl_seconds: 60 # SQL 型PostgreSQL registry: registry_type: sql path: postgresql://user:passhost:5432/feast cache_ttl_seconds: 60 # 远程 Registry registry: registry_type: remote path: grpc://feast-registry-server:6570在线存储Online Store类型关键配置适用场景sqlitepath本地开发redisconnection_string生产环境、低延迟dynamodbregionAWS 原生datastoreproject_idGCP 原生bigtableproject_id、instanceGCP 高吞吐postgreshost、port、database、user、password自管理部署snowflake.onlineaccount、database、schemaSnowflake 生态milvushost、port向量检索qdranthost、port向量检索remotepath远程特征服务# SQLite本地开发 online_store: type: sqlite path: data/online_store.db # Redis生产 online_store: type: redis connection_string: redis://localhost:6379 # PostgreSQL online_store: type: postgres host: localhost port: 5432 database: feast db_schema: public user: postgres password: secret # Milvus向量检索 online_store: type: milvus host: localhost port: 19530各在线存储的实现位于 sdk/python/feast/infra/online_stores 目录可按需查看RedisOnlineStore、DynamoDBOnlineStore等类的连接与读写逻辑。离线存储Offline Store类型适用场景dask本地开发默认duckdb本地、快速分析bigqueryGCPsnowflake.offlineSnowflakeredshiftAWSspark大规模处理postgres自管理部署trino联邦查询athenaAWS Serverlessclickhouse分析场景remote远程离线服务# DuckDB offline_store: type: duckdb # BigQuery offline_store: type: bigquery project_id: my-gcp-project dataset: feast_dataset # Spark offline_store: type: spark spark_conf: spark.master: local[*]认证、特征服务与可观测性认证类型包括no_auth默认、kubernetesKubernetes 服务账号、oidc服务端 OpenID Connect、oidc_client客户端 OIDCauth: type: oidc client_id: feast-client auth_server_url: https://auth.example.com auth_discovery_url: https://auth.example.com/.well-known/openid-configuration特征服务器可通过feature_server.type配置local启动本地 Python 特征服务器mcp启动基于 MCP 的特征服务器。物化行为可通过materialization.pull_latest_features控制设为false时每个实体只拉取最新特征值。如需数据血缘追踪可开启 OpenLineageopenlineage: enabled: true transport_type: http # http、console、file、kafka省略则用 OpenLineage SDK 默认 transport_url: http://feast-example-lineage:6580 transport_endpoint: api/v1/lineage namespace: feast emit_on_apply: true emit_on_materialize: true特征仓库目录布局与注册机制my_feature_repo/ ├── feature_store.yaml # 必需配置 ├── .feastignore # 可选类 gitignore 的忽略文件 ├── driver_features.py # 特征定义 ├── customer_features.py # 更多定义 └── data/ ├── driver_stats.parquet # 数据文件供 FileSource 使用 └── registry.db # 自动生成的 Registry要点feast apply会递归扫描仓库内所有.py文件发现特征定义扫描与注册逻辑见 sdk/python/feast/repo_operations.py因此多个 Python 文件可自由拆分组织用.feastignore可排除无需扫描的文件/目录配置文件的完整字段说明可参考 docs/reference/feature-store-yaml.md 与 docs/getting-started/concepts/feature-repo.md。FeatureStore API 速查方法用途apply(objects)注册实体、特征视图、ODFV、SFV、特征服务、数据源plan(desired_registry)预览 apply 变更get_online_features(features, entity_rows)在线低延迟查询get_historical_features(entity_df, features)时间点正确连接的历史训练数据materialize(start_date, end_date)离线 → 在线全量物化materialize_incremental(end_date)增量物化push(push_source_name, df)推送数据到在线/离线存储write_to_online_store(fv_name, df)直写在线存储write_to_offline_store(fv_name, df)直写离线存储retrieve_online_documents(feature, query, top_k)向量相似度检索v1retrieve_online_documents_v2(...)向量检索 v2文本/向量/图像list_entities()/list_feature_views()列出实体 / 特征视图list_on_demand_feature_views()/list_stream_feature_views()列出 ODFV / SFVlist_feature_services()/list_data_sources()列出特征服务 / 数据源get_entity(name)/get_feature_view(name)按名获取实体 / 特征视图get_feature_service(name)按名获取特征服务delete_feature_view(name)/delete_feature_service(name)删除特征视图 / 特征服务create_saved_dataset(...)保存数据集以供复用refresh_registry()强制刷新 Registry 缓存teardown()清理全部基础设施资源serve(port)/serve_ui(port)启动特征服务器 / Feast UIserve_registry(port)/serve_offline(port)启动 Registry 服务器 / 离线服务器常用导入清单from feast import ( Entity, FeatureView, OnDemandFeatureView, FeatureService, Field, FileSource, RequestSource, FeatureStore, ) from feast.on_demand_feature_view import on_demand_feature_view from feast.types import Float32, Float64, Int64, String, Bool, Array from feast.value_type import ValueType from datetime import timedelta深入阅读本文对应的三个参考文档存放于 skills/references 目录可按需深入特征定义全量参考所有类型、参数与写法模式skills/references/feature-definitions.md配置参考feature_store.yaml、各类存储、认证skills/references/configuration.md检索与 RAG 参考在线/离线检索、向量搜索、RAG Retrieverskills/references/retrieval-and-rag.md。此外仓库中还有配套的可运行示例可以直接对照学习最简本地仓库见 examples/quickstart/quickstart.ipynb端到端特征仓库见 examples/credit-risk-end-to-endRAG 检索实践见 examples/rag-retriever/rag_feast.ipynb 与 examples/rag-docling/docling-quickstart.ipynb。【免费下载链接】feastThe Open Source Feature Store for AI/ML项目地址: https://gitcode.com/GitHub_Trending/fe/feast创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考