Feast 第三方集成生态插件标准、扩展机制与自定义 Store 开发指南【免费下载链接】feastThe Open Source Feature Store for AI/ML项目地址: https://gitcode.com/GitHub_Trending/fe/feastFeast 作为开源 Feature Store其核心设计目标之一就是融入你已有的技术栈——无论是数据源BigQuery、Snowflake、Redshift、Kafka还是离线/在线存储Redis、DynamoDB、DuckDB、向量数据库Feast 都提供了官方内置或社区维护的插件化集成。本文以 Feast 仓库的第三方集成文档为主线系统梳理集成清单与 Roadmap、插件被收录和合入主仓库必须满足的工程标准并结合仓库源码深入讲解自定义离线存储Offline Store与在线存储Online Store的接口实现、配置加载与通用测试机制帮助你判断我的存储/数据源如何接入 Feast以及如何把一个插件从 contrib 提升为官方集成。一、Feast 的第三方集成生态概览Feast 与一大批工具和技术集成覆盖了数据源、离线存储、在线存储、向量检索、流式摄取、部署形态、特征服务与治理等完整链路。这些集成大部分以插件plugin的形式维护在 Feast 主仓库内sdk/python/feast/infra/下少部分由社区在独立仓库维护。官方对集成的完整功能清单与演进状态集中在 docs/roadmap.md 中其中每一项都标注了完成状态[x]已完成 /[ ]规划中以及对应的参考文档链接。从 roadmap 的目录结构可以清晰地看到 Feast 集成面的广度分类已完成的代表性集成数据源Snowflake、Redshift、BigQuery、Parquet 文件、Kafka/Kinesis经由 push 支持、Postgres、Spark、Athena、ClickHouse、Oracle、MongoDB、Ray 等离线存储Snowflake、Redshift、BigQuery、DuckDB、Dask、Remote、Hybrid以及 MSSQL、Hive、Trino、Spark、Couchbase、ClickHouse、Ray、Oracle、MongoDB 等 contrib 插件在线存储Snowflake、DynamoDB、Redis、Dragonfly、Datastore、Bigtable、SQLite、Remote、Postgres、HBase、Cassandra/AstraDB、ScyllaDB、MySQL、Hazelcast、Elasticsearch、SingleStore、Couchbase、MongoDB、Aerospike向量存储Qdrant、Milvus、Faiss作为向量在线存储流式自定义流式摄取作业、基于 push 的流式数据写入在线/离线存储部署AWS Lambda、Kubernetes见 docs/how-to-guides/running-feast-in-production.md特征服务Python 客户端、Python Feature Server、Java/Go Feature Serveralpha、Offline Feature Server、Registry Server、Feast Operator治理与发现Python SDK / CLI 浏览 registry、Feature Service、Amundsen 与 DataHub 集成、Feast Web UI需要特别说明的是这份清单不是封闭的。如果你在列表中没有找到自己正在使用的离线存储或在线存储Feast 明确提供了两条自己动手的路径添加一个新的离线存储添加一个新的在线存储二、插件Plugin的定位与 contrib 机制在深入自定义开发之前先理解 Feast 对插件的等级划分。从 sdk/python/feast/infra/offline_stores/ 与 sdk/python/feast/infra/online_stores/ 的目录结构可以直观看到官方内置实现如bigquery.py、redshift.py、snowflake.py、redis.py、sqlite.py、dynamodb.py直接位于父目录下而社区贡献的实现统一放在contrib/子目录中例如离线存储 contribathena_offline_store、clickhouse_offline_store、mssql_offline_store、oracle_offline_store、postgres_offline_store、ray_offline_store、spark_offline_store、trino_offline_store、couchbase_offline_store、mongodb_offline_store在线存储 contrib在线存储目录中按子包组织mysql_online_store、hbase_online_store、cassandra_online_store、hazelcast_online_store、elasticsearch_online_store、singlestore_online_store、qdrant_online_store、milvus_online_store、aerospike_online_store、scylladb_online_store等contrib 插件的定位官方明确声明不保证实现接口的全部方法不保证 API 稳定后续可能发生变化实现中应通过RuntimeWarning向用户提示这是一个 contrib 插件不由 Feast 维护者维护源码中的标准警告文案为This offline store is an experimental feature in alpha development. Some functionality may still be unstable so functionality can change in the future.从 contrib 提升为官方插件需要满足两个条件CImake test-python-integration已配置为对该插件运行全部测试并通过至少两名贡献者拥有该插件的维护权理想情况下记录在OWNERS/CODEOWNERS文件中。三、集成被收录与合入主仓库的工程标准第三方集成文档的核心价值在于定义了清晰的验收门槛分为两个层级3.1 插件被收录被官方列出的要求一个插件集成若要被官方收录并突出展示必须满足以下三条必须有测试。理想情况下应复用 Feast 的 universal 测试套件示例见 添加或复用测试自定义测试也可以接受必须有基本的使用文档说明如何使用该插件作者必须与维护者协作通过一次基础代码评审确保实现与 Feast 核心实现大致对齐例如遵循相同的数据模型与调用约定。3.2 插件合入主仓库的要求若插件要合并进 Feast 主仓库标准更严格PR 必须通过全部集成测试且必须更新 universal 测试专门为自定义集成设计的测试以覆盖该集成必须有文档和教程说明如何使用该集成作者或其他贡献者必须承诺拥有这些文件的维护权并持续维护如果插件由组织而非个人贡献该组织应为集成测试提供基础设施或云额度。这套标准保证了第三方集成不仅仅是能跑而是可持续、可验证、有归属的工程资产。四、自定义离线存储接口、配置与调用链离线存储是 Feast 中读历史特征、做 point-in-time join、支撑物化的存储与计算层。完整的自定义教程见 添加一个新的离线存储其流程分为 8 步定义OfflineStore类、定义OfflineStoreConfig类、定义RetrievalJob类、定义DataSource类、在feature_store.yaml中引用、测试、更新依赖、补充文档。4.1 OfflineStore 接口离线存储的抽象基类定义在 sdk/python/feast/infra/offline_stores/offline_store.py 中OfflineStore类OfflineStore(ABC)。从源码看一个完整的离线存储实现需要覆盖以下核心方法方法触发时机说明pull_latest_from_table_or_queryfeast materialize/feast materialize-incremental命令或FeatureStore.materialize()从离线存储拉取数据由FeatureStore负责写入在线存储get_historical_featuresFeatureStore.get_historical_features()读取历史特征典型用于训练 ML 模型执行 point-in-time 正确性连接pull_all_from_table_or_query可选SavedDatasets 与特征质量监控的兜底计算路径按起止日期拉取全部数据write_logged_features可选内部用于 SavedDatasets将 pyarrow 表或 parquet 路径写入LoggingSource/LoggingConfig指定的源offline_write_batch可选Push API将 pyarrow 表直接推送到指定 feature view 的 batch source接口签名示例节选自自定义教程def get_historical_features(self, config: RepoConfig, feature_views: List[FeatureView], feature_refs: List[str], entity_df: Union[pd.DataFrame, str], registry: Registry, project: str, full_feature_names: bool False) - RetrievalJob: Perform point-in-time correct join of features onto an entity dataframe. # Implementation here. pass一个值得注意的实现细节OfflineStore类名必须以OfflineStore后缀结尾教程中以 hint 形式强调。这一约定不是装饰性的——sdk/python/feast/repo_config.py 中的get_offline_store_type函数在加载时会校验如果配置的 type 不在OFFLINE_STORE_CLASS_FOR_TYPE字典中则要求字符串以OfflineStore结尾否则抛出FeastOfflineStoreInvalidName。同时Feast 通过OFFLINE_STORE_CLASS_FOR_TYPE字典如file、bigquery、redshift、duckdb等实现了简短别名 → 全限定类名的映射第三方实现与第一方实现走同一条类加载代码路径。另外与 OnlineStore 不同Feast 不为离线存储管理任何基础设施——离线存储本质上是计算 存储的查询引擎其方法如get_historical_features、pull_latest_from_table_or_query均为静态方法直接对底层引擎发起查询。4.2 RetrievalJob惰性执行的查询句柄离线存储的方法不应急切执行读操作而是返回一个RetrievalJob实例代表对底层存储的实际查询执行。RetrievalJob抽象类同样定义在 sdk/python/feast/infra/offline_stores/offline_store.py 中其核心能力包括to_df()同步执行查询并返回 pandas DataFrame会执行 on-demand 变换与可选的验证to_arrow()同步执行查询并返回 pyarrow Tableto_df内部实际委托给to_arrow().to_pandas()to_remote_storage()可选将结果以多个 parquet 文件导出到远端存储如 S3/GCS配合自定义 Materialization Engine 实现可扩展的批量物化——将物化数据分块并行处理。若未实现Feast 默认走本地物化把全部记录拉入内存persist()、metadata、to_sql()等辅助能力。自定义RetrievalJob的典型实现模式以自定义文件离线存储为例class CustomFileRetrievalJob(RetrievalJob): def __init__(self, evaluation_function: Callable): Initialize a lazy historical retrieval job # The evaluation function executes a stored procedure to compute a historical retrieval. self.evaluation_function evaluation_function def to_df(self): # Only execute the evaluation function to build the final historical retrieval dataframe at the last moment. df self.evaluation_function() return df def to_arrow(self): df self.evaluation_function() return pyarrow.Table.from_pandas(df) def to_remote_storage(self): # Optional method to write to an offline storage location to support scalable batch materialization. pass4.3 OfflineStoreConfig基于 pydantic 的配置模型所有OfflineStore实现必须在同文件内定义对应的OfflineStoreConfig类该类继承自 sdk/python/feast/repo_config.py 中的FeastConfigBaseModel一个 pydantic 模型负责把 YAML 配置解析为 Python 对象并可通过 validator 校验配置正确性。命名约束是配置类名 离线存储类名 Config后缀且必须包含type字段值为对应OfflineStore类的全限定类名。class CustomFileOfflineStoreConfig(FeastConfigBaseModel): Custom offline store config for local (file-based) store type: Literal[feast_custom_offline_store.file.CustomFileOfflineStore] \ feast_custom_offline_store.file.CustomFileOfflineStore uri: str # URI for your offline store(in this case it would be a path)对应在feature_store.yaml中的配置方式project: my_project registry: data/registry.db provider: local offline_store: type: feast_custom_offline_store.file.CustomFileOfflineStore uri: File URI online_store: path: data/online_store.db该配置会以config.offline_store字段的形式注入到OfflineStore的每个方法参数config: RepoConfig中实现中通过assert isinstance(offline_store_config, CustomFileOfflineStoreConfig)取回类型安全的配置对象。4.4 DataSource 与类型映射要让自定义离线存储作为 feature view 的 batch source 使用还需定义DataSource的子类实现from_proto与to_proto两个方法。对于不在主仓库实现的自定义离线存储应使用custom_options字段存储数据源所需配置并在to_proto中将其序列化为字节、在from_proto中读回通常以 JSON 形式编码class CustomFileDataSource(FileSource): Custom data source class for local files staticmethod def from_proto(data_source: DataSourceProto): custom_source_options str( data_source.custom_options.configuration, encodingutf8 ) path json.loads(custom_source_options)[path] return CustomFileDataSource( field_mappingdict(data_source.field_mapping), pathpath, timestamp_fielddata_source.timestamp_field, created_timestamp_columndata_source.created_timestamp_column, date_partition_columndata_source.date_partition_column, ) def to_proto(self) - DataSourceProto: config_json json.dumps({path: self.path}) data_source_proto DataSourceProto( typeDataSourceProto.CUSTOM_SOURCE, custom_optionsDataSourceProto.CustomSourceOptions( configurationbytes(config_json, encodingutf8) ), ) data_source_proto.timestamp_field self.timestamp_field data_source_proto.created_timestamp_column self.created_timestamp_column data_source_proto.date_partition_column self.date_partition_column return data_source_proto此外大多数离线存储都需要实现自定义类型映射在DataSource类中实现source_datatype_to_feast_value_type将数据源类型转换为 Feast 值类型与get_column_names_and_types获取列名与对应类型。类型转换的辅助函数可以放入sdk/python/feast/type_map.py。类型映射必须正确否则 Feast 处理特征列时可能发生错误转换导致信息丢失或数据错误——这是自定义 store 开发中出错率最高的环节之一。4.5 在 feature repo 中使用自定义离线存储只要OfflineStore类在 Python 环境中可用Feast 会在运行时动态导入。feature_store.yaml中offline_store.type应指定为可导入的全限定类名project: test_custom registry: data/registry.db provider: local offline_store: # Make sure to specify the type as the fully qualified path that Feast can import. type: feast_custom_offline_store.file.CustomFileOfflineStore如果无需额外配置可以省略其他字段只写typeoffline_store: feast_custom_offline_store.file.CustomFileOfflineStore随后在 repo 定义文件中使用自定义数据源定义 FeatureViewdriver_hourly_stats CustomFileDataSource( pathfeature_repo/data/driver_stats.parquet, timestamp_fieldevent_timestamp, created_timestamp_columncreated, ) driver_hourly_stats_view FeatureView( sourcedriver_hourly_stats, ... )五、自定义在线存储基础设施管理与读写方法在线存储是服务阶段低延迟读取特征值的存储层。完整教程见 添加一个新的在线存储流程为 6 步定义OnlineStore类、定义OnlineStoreConfig类、在feature_store.yaml中引用、测试、更新依赖、补充文档。5.1 OnlineStore 接口的两组方法抽象基类定义在 sdk/python/feast/infra/online_stores/online_store.pyOnlineStore(ABC)。从源码结构看其方法分为两组1基础设施管理方法——与离线存储不同Feast会为在线存储管理基础设施方法触发时机职责updatefeast apply命令或FeatureStore.apply()SDK 方法执行写/读数据前的必要操作例如为新的 feature view 创建 MySQL 表与索引teardownfeast teardown或FeatureStore.teardown()执行清理例如删除被删 feature view 对应的表与索引2读写方法方法触发时机说明online_write_batchfeast materialize/feast materialize-incremental或FeatureStore.materialize()批量写入特征行数据以(EntityKeyProto, Dict[str, ValueProto], event_ts, created_ts)四元组列表传入online_readFeatureStore.get_online_features()按实体键读取特征值返回与entity_keys等长的(event_ts, {feature_name: ValueProto})列表在线存储实现中实体键需要经过serialize_entity_key(entity_key, entity_key_serialization_versionconfig.entity_key_serialization_version)序列化后写入存储读取时通过ValueProto.ParseFromString()还原特征值protobuf 编码。此外OnlineStore基类还提供get_online_features、plan、异步读写、预计算向量读写等扩展能力自定义实现可根据需要覆盖。5.2 OnlineStoreConfig 与 feature_store.yaml 引用与离线存储相同的约定OnlineStoreConfig类继承FeastConfigBaseModel、以Config后缀命名、必须含type字段。以 MySQL 为例class MySQLOnlineStoreConfig(FeastConfigBaseModel): type: Literal[feast_custom_online_store.mysql.MySQLOnlineStore] feast_custom_online_store.mysql.MySQLOnlineStore host: Optional[StrictStr] None user: Optional[StrictStr] None password: Optional[StrictStr] None database: Optional[StrictStr] Noneproject: test_custom registry: data/registry.db provider: local online_store: # Make sure to specify the type as the fully qualified path that Feast can import. type: feast_custom_online_store.mysql.MySQLOnlineStore user: foo password: bar同样若无需额外配置可以简写为online_store: feast_custom_online_store.mysql.MySQLOnlineStore。配置在方法内通过config.online_store访问例如hostonline_store_config.host or 127.0.0.1。类名必须以OnlineStore结尾repo_config.py中的get_online_config_from_type会据此校验并动态加载。六、插件的通用测试机制FULL_REPO_CONFIGS 与 universal 测试Feast 为第三方存储插件提供了复用官方测试套件的能力这是插件质量门槛必须有测试的技术支撑。测试体系细节见 添加或复用测试。6.1 测试套件结构单元测试位于sdk/python/tests/unit集成测试位于sdk/python/tests/integration集成测试按组件组织e2e端到端、offline_store历史检索、push API、特征日志、online_store在线检索、push、registrationregistry、CLI、类型推断、materialization、feature_repos测试夹具与配置等universal feature repo是一套可参数化的夹具如environment与universal_data_sources可覆盖不同离线存储 × 在线存储 × provider 的组合让同一测试代码跑遍所有存储后端测试标记markerpytest.mark.integration标记集成测试pytest.mark.universal_offline_stores将测试参数化到所有 universal 离线存储file、redshift、bigquery、snowflakeuniversal_online_stores(only[redis])可指定特定在线存储。6.2 用 FULL_REPO_CONFIGS_MODULE 覆盖测试矩阵核心机制在 sdk/python/tests/universal/feature_repos/repo_configuration.pyFeast 通过FULL_REPO_CONFIGS变量参数化集成测试该变量聚合AVAILABLE_OFFLINE_STORES每个条目为(provider, DataSourceCreator)与AVAILABLE_ONLINE_STORES每个条目为(store_config, OnlineStoreCreator)生成IntegrationTestRepoConfig插件无需修改 Feast 仓库即可覆盖测试矩阵创建自己的模块文件定义AVAILABLE_OFFLINE_STORES/AVAILABLE_ONLINE_STORES/FULL_REPO_CONFIGS通过环境变量FULL_REPO_CONFIGS_MODULE指向该模块再用make test-python-universal运行。仓库中真实的 contrib 插件配置示例 sdk/python/feast/infra/offline_stores/contrib/postgres_repo_configuration.pyfrom feast.infra.offline_stores.contrib.postgres_offline_store.tests.data_source import ( PostgreSQLDataSourceCreator, ) from tests.universal.feature_repos.repo_configuration import REDIS_CONFIG from tests.universal.feature_repos.universal.online_store.redis import ( RedisOnlineStoreCreator, ) AVAILABLE_OFFLINE_STORES [(local, PostgreSQLDataSourceCreator)] AVAILABLE_ONLINE_STORES {redis: (REDIS_CONFIG, RedisOnlineStoreCreator)}配套的 Makefile 目标以 Spark 为例摘自自定义离线存储教程test-python-universal-spark: PYTHONPATH. \ FULL_REPO_CONFIGS_MODULEsdk.python.feast.infra.offline_stores.contrib.spark_repo_configuration \ PYTEST_PLUGINSfeast.infra.offline_stores.contrib.spark_offline_store.tests \ IS_TESTTrue \ python -m pytest -n 8 --integration \ -k not test_historical_retrieval_fails_on_validation and \ not test_historical_retrieval_with_validation and ... \ sdk/python/tests其中PYTEST_PLUGINS环境变量让 pytest 加载DataSourceCreator/OnlineStoreCreator-k选项可剔除与特定存储无关或暂不适用的测试。6.3 DataSourceCreator 与 OnlineStoreCreatorDataSourceCreator.create_data_source(df, ...)把测试传入的 DataFrame 上传/写入对应离线存储返回指向该位置的数据源对象BigQueryDataSourceCreator是参考实现OnlineStoreCreator用于在容器中拉起在线存储实例。例如 Redis 的 creator 会启动容器并等待 Ready to accept connections 日志后再开始测试这样其他开发者无需自建实例即可运行测试。若集成测试失败说明该存储插件的实现存在错误——这是最直接的信号。测试通过后还应把数据源注册到repo_config.py的OFFLINE_STORE_CLASS_FOR_TYPE/ONLINE_STORE_CLASS_FOR_TYPE字典中与spark、trino等一致这样 Feast 才能从feature_store.yaml中加载对应类。七、依赖、文档与合入流程的收尾一个插件在代码与测试就绪后还需完成以下工程化步骤依赖管理在 sdk/python/pyproject.toml 的[project.optional-dependencies]下为插件新增 extra如offline_store [package11.0, package2]使其默认不随 Feast 安装随后运行make lock-python-dependencies-all重新生成锁文件文档离线存储在docs/reference/offline-stores/与docs/reference/data-sources/各新增一个 Markdown 文件并在docs/reference/data-sources/README.md与docs/SUMMARY.md中登记索引在线存储在docs/reference/online-stores/新增文档并在其README.md与docs/SUMMARY.md中登记文档必须覆盖如何创建数据源、feature_store.yaml需要哪些配置、明确标注该数据源处于 alpha 开发阶段、该存储的数据模型说明最后运行make build-sphinx生成 Python API 文档合入主仓库前的收尾按前文合入主仓库的四条要求逐一核对——PR 通过全部集成测试并更新 universal 测试、提供文档与教程、明确文件所有权与后续维护承诺、组织贡献者提供测试基础设施或额度。八、总结从接入到贡献的完整路径Feast 的第三方集成机制可以总结为一条清晰的路径查清单在 docs/roadmap.md 确认你的技术栈是否已有集成以及其状态官方 / contrib / 规划中选路线没有现成集成时按 添加离线存储 或 添加在线存储 教程实现接口注意类名后缀、Config 类命名、type字段与类型映射等硬性约定接入使用在feature_store.yaml中用全限定类名引用Feast 运行时动态导入过测试通过FULL_REPO_CONFIGS_MODULEmake test-python-universal复用 universal 测试套件用DataSourceCreator/OnlineStoreCreator接入测试基础设施工程化收尾以 extra 形式管理依赖、补齐参考文档与教程、登记索引最后对照收录三条 合入四条标准推进评审与合入。理解这套标准与机制既能让现有技术栈顺利接入 Feast也能为贡献者提供一份可执行、可验收的插件开发路线图——这正是第三方集成文档的核心价值所在。【免费下载链接】feastThe Open Source Feature Store for AI/ML项目地址: https://gitcode.com/GitHub_Trending/fe/feast创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考