Feast Push Source 完全指南实时推送特征到在线/离线存储【免费下载链接】feastThe Open Source Feature Store for AI/ML项目地址: https://gitcode.com/GitHub_Trending/fe/feastPush Source 是 Feast 提供的一种按需注入型数据源允许应用在运行时把特征值实时推送到在线存储Online Store与离线存储Offline Store从而让在线服务立刻使用到最新鲜的特征。本文以 docs/reference/data-sources/push.md 为骨架结合 data_source.py 与 feature_store.py 的源码实现完整讲解 Push Source 的定义方式、推送模式PushMode、与流式 FeatureView 的配合、Spark Streaming 集成以及 Python Feature Server 的 HTTP 推送接口帮助你从定义到生产部署一次掌握。Push Source 是什么Push Source 是 Feast 中一类特殊的数据源它不像 BigQuery、Redshift 那样绑定一张物理表或一个流式系统而是一个接收端客户端通过 SDK 或 HTTP 接口把一批数据通常是一个 pandas DataFrame直接交到 Feast 手里Feast 再把这批数据写入在线存储或离线存储。其核心价值体现在两点实时性特征值被推入后即可立即供在线推理读取无需等待离线任务周期性地物化materialization。统一传播一个 Push Source 可以被多个 Feature View 共用。当数据被推送到某个 Push Source 时Feast 会把特征值传播给所有以它为流式源stream source的 Feature View一次推送、多处生效。这一机制在源码中清晰可见。FeatureStore.push见 feature_store.py首先通过_fvs_for_push_source_or_raisefeature_store.py从注册表中找出所有stream_source是PushSource且名字匹配的 Feature View 与 Stream Feature View再逐一调用write_to_online_store/write_to_offline_store完成写入for fv in self._fvs_for_push_source_or_raise(push_source_name, allow_registry_cache): if to PushMode.ONLINE or to PushMode.ONLINE_AND_OFFLINE: self.write_to_online_store(fv.name, df, allow_registry_cacheallow_registry_cache, transform_on_writetransform_on_write) pushed_fv_names.append(fv.name) if to PushMode.OFFLINE or to PushMode.ONLINE_AND_OFFLINE: self.write_to_offline_store(fv.name, df, allow_registry_cacheallow_registry_cache)若没有找到任何消费该 Push Source 的 Feature View会抛出PushSourceNotFoundException定义于 errors.py提示你检查 Push Source 名称是否拼写正确、是否已被 Feature View 引用。与 write_to_online_store 的关系Push Source 是对旧的FeatureStore.write_to_online_storeAPI 的取代方案supersede。旧 API 直接以 Feature View 为操作对象需要你显式指定 Feature View 名称而push以数据源为中心Feast 自动推导出所有受影响的 Feature View语义更清晰、扩展性更好。旧接口仍保留用于兼容例如 feature_server.py 中/write-to-online-store端点仍调用它但新项目应优先使用 Push Source。是否需要 batch_sourcePush Source 可以可选地指定一个batch_source作为它的离线底表指定了 batch_source可以用于检索历史特征训练数据并且支持从离线存储物化到在线存储这一标准流程未指定 batch_source适用于特征只在训练之后生成、或者只在线使用例如向量数据库中的 embedding的场景。关于 batch_source 还有两点需要注意当你使用 batch_source 时你有责任保证数据也被写入该批量数据源如数据仓库即推送 写底表双写由你自己维护当 Push Source 被用作 Feature View 的流式源时Feature View 自身不需要再显式指定 batch_source——Feature View 会从 Push Source 继承 batch_source 信息。这一点也体现在序列化逻辑中PushSource._to_proto_impldata_source.py会把 batch_source 的timestamp_field、created_timestamp_column、field_mapping、date_partition_column继承到 Push Source 的 proto 中方便下游 Feature View 复用。典型流式数据架构流式数据是 Push Source 最重要的应用场景。一个典型的流式特征流水线如下原始事件到达流 1施加流式变换生成特征例如last_N_purchased_categories等聚合特征流 2可选把流 2 的值写入离线存储作为用于训练的历史日志把流 2 的值写入在线存储用于低延迟的特征服务周期性从离线存储物化特征到在线存储缩小训练与服务之间的偏差training-serving skew提升模型性能。Feast 同时支持两种推送方向把已在 Feature View 中注册的特征推送到在线存储以获得更新鲜的特征也可以指定推送到离线存储将一批流数据写入初始化 Feature Store 时仓库配置feature_store.yaml中声明的离线存储。基础示例定义 Push Source 并推送数据定义 Push Source下面的示例完整演示了如何定义一个以 BigQuery 表为底表的 Push Source并挂载到 Feature View 上。注意Push Source 的 schema 中必须包含实体列entity。from feast import Entity, PushSource, ValueType, BigQuerySource, FeatureView, Feature, Field from feast.types import Int64 push_source PushSource( namepush_source, batch_sourceBigQuerySource(tabletest.test), ) user Entity(nameuser, join_keys[user_id]) fv FeatureView( namefeature view, entities[user], schema[Field(namelife_time_value, dtypeInt64)], sourcepush_source, )对应到源码PushSource.__init__data_source.py只接收name、batch_source、description、tags、owner五个参数其中name是必填项batch_source为可选的DataSource类型。它继承自DataSource基类因此也具备timestamp_field、created_timestamp_column、field_mapping等通用数据源属性可从 batch_source 自动继承。推送数据推送时通过to参数控制目标存储它接受三种模式定义于 data_source.py 的PushMode枚举PushMode值行为PushMode.ONLINE1只写入在线存储默认值PushMode.OFFLINE2只写入离线存储PushMode.ONLINE_AND_OFFLINE3同时写入在线与离线存储to参数是可选的默认只推送到在线存储from feast import FeatureStore import pandas as pd from feast.data_source import PushMode fs FeatureStore(...) feature_data_frame pd.DataFrame() fs.push(push_source_name, feature_data_frame, toPushMode.ONLINE_AND_OFFLINE)FeatureStore.push的完整签名还包含allow_registry_cache是否允许使用注册表缓存默认 True与transform_on_write写入前是否执行变换默认 True两个参数见 feature_store.py。推送到离线存储时的列校验当to包含离线存储时会走write_to_offline_storefeature_store.py这里有一个容易踩坑的细节输入 DataFrame 的列必须与 Feature View 的 batch_source 列完全一致。源码会先取 batch_source 的列集合与输入 DataFrame 列集合做对比不一致时抛出 ValueError明确指出缺失列与多余列if input_columns_set ! source_columns_set: missing_expected_columns sorted(source_columns_set - input_columns_set) extra_unexpected_columns sorted(input_columns_set - source_columns_set) raise ValueError(...)reorder_columnsTrue默认时Feast 还会自动把 DataFrame 的列重排为与源表一致然后再通过provider.ingest_df_to_offline_store写入。Spark Streaming 集成示例在 PySpark 场景下最直接的做法是把 Feast Python SDK 引入已有的 PySpark 流水线在foreachBatch中把 Spark DataFrame 转为 pandas DataFrame 后推送from feast import FeatureStore store FeatureStore(...) spark SparkSession.builder.getOrCreate() streamingDF spark.readStream.format(...).load() def feast_writer(spark_df): pandas_df spark_df.to_pandas() store.push(driver_hourly_stats, pandas_df) streamingDF.writeStream.foreachBatch(feast_writer).start()这个模式可以被 contrib 流处理器在底层复用。仓库中与之对应的一组集成测试验证了端到端行为test_push_features_to_online_store.py验证推送到在线存储后能被get_online_features读取test_push_features_to_offline_store.py验证推送到离线存储后能用于历史检索test_stream_feature_view.py验证 StreamFeatureView 与 Push Source 的注册与解析。通过 Python Feature Server 推送在部署了 Python Feature Server 的场景下无需引入 SDK 即可通过 HTTP 接口推送数据。Feature Server 暴露了POST /push端点见 feature_server.py请求体为PushFeaturesRequest其中包含push_source_name、dfDataFrame 序列化数据、toonline/offline/online_and_offline、allow_registry_cache、transform_on_write等字段。to字段的解析映射与 SDK 的PushMode一一对应if request.to offline: to PushMode.OFFLINE elif request.to online: to PushMode.ONLINE elif request.to online_and_offline: to PushMode.ONLINE_AND_OFFLINE else: raise ValueError(...请指定 [online, offline, online_and_offline] 之一)服务端会先校验调用者的 RBAC 权限分别对应WRITE_ONLINE、WRITE_OFFLINE动作再根据 provider 是否支持异步在线写入决定走store.push_async还是线程池中的同步store.push。需要注意离线存储写入目前只有同步实现异步仅适用于在线存储部分见 feature_server.py 中的注释说明。此外若启用了离线推送批处理offline_push_batching_enabled见 feature_server.py离线推送会被OfflineWriteBatcher缓冲并按批写入此时 HTTP 响应码为 202 Accepted。完整部署与调用方式参见 Python Feature Server 文档。进阶StreamFeatureView 与构建流式特征Push Source 是构建StreamFeatureView的基础两者结合可以实现流式变换 → 推送 → 实时服务的完整链路。流式特征构建的完整教程包括示例项目、数据准备与端到端演示见 Tutorial: Building streaming features。在仓库的示例与测试仓库中也能看到典型的 Push Source StreamFeatureView 组合写法例如 feature_views.py 中的通用测试特征视图以及 example_feature_repo_1.py 中的示例特征仓库可作为定义 Feature View、Entity 与 Push Source 时的参考模板。小结Push Source 让特征值可以按需实时注入在线/离线存储取代旧的write_to_online_storeAPI一个 Push Source 可被多个 Feature View 共享一次推送自动传播到所有消费方batch_source可选有它可支持历史检索与物化无它适用于仅在线场景如 embedding推送方向由PushModeONLINE / OFFLINE / ONLINE_AND_OFFLINE控制默认仅在线生产环境可通过 Python Feature Server 的POST /push接口含 RBAC 权限校验与可选的离线批处理完成推送Spark Streaming 场景下在foreachBatch中调用store.push即可与现有 PySpark 流水线无缝集成。【免费下载链接】feastThe Open Source Feature Store for AI/ML项目地址: https://gitcode.com/GitHub_Trending/fe/feast创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Feast Push Source 完全指南：实时推送特征到在线/离线存储
Feast Push Source 完全指南实时推送特征到在线/离线存储【免费下载链接】feastThe Open Source Feature Store for AI/ML项目地址: https://gitcode.com/GitHub_Trending/fe/feastPush Source 是 Feast 提供的一种按需注入型数据源允许应用在运行时把特征值实时推送到在线存储Online Store与离线存储Offline Store从而让在线服务立刻使用到最新鲜的特征。本文以 docs/reference/data-sources/push.md 为骨架结合 data_source.py 与 feature_store.py 的源码实现完整讲解 Push Source 的定义方式、推送模式PushMode、与流式 FeatureView 的配合、Spark Streaming 集成以及 Python Feature Server 的 HTTP 推送接口帮助你从定义到生产部署一次掌握。Push Source 是什么Push Source 是 Feast 中一类特殊的数据源它不像 BigQuery、Redshift 那样绑定一张物理表或一个流式系统而是一个接收端客户端通过 SDK 或 HTTP 接口把一批数据通常是一个 pandas DataFrame直接交到 Feast 手里Feast 再把这批数据写入在线存储或离线存储。其核心价值体现在两点实时性特征值被推入后即可立即供在线推理读取无需等待离线任务周期性地物化materialization。统一传播一个 Push Source 可以被多个 Feature View 共用。当数据被推送到某个 Push Source 时Feast 会把特征值传播给所有以它为流式源stream source的 Feature View一次推送、多处生效。这一机制在源码中清晰可见。FeatureStore.push见 feature_store.py首先通过_fvs_for_push_source_or_raisefeature_store.py从注册表中找出所有stream_source是PushSource且名字匹配的 Feature View 与 Stream Feature View再逐一调用write_to_online_store/write_to_offline_store完成写入for fv in self._fvs_for_push_source_or_raise(push_source_name, allow_registry_cache): if to PushMode.ONLINE or to PushMode.ONLINE_AND_OFFLINE: self.write_to_online_store(fv.name, df, allow_registry_cacheallow_registry_cache, transform_on_writetransform_on_write) pushed_fv_names.append(fv.name) if to PushMode.OFFLINE or to PushMode.ONLINE_AND_OFFLINE: self.write_to_offline_store(fv.name, df, allow_registry_cacheallow_registry_cache)若没有找到任何消费该 Push Source 的 Feature View会抛出PushSourceNotFoundException定义于 errors.py提示你检查 Push Source 名称是否拼写正确、是否已被 Feature View 引用。与 write_to_online_store 的关系Push Source 是对旧的FeatureStore.write_to_online_storeAPI 的取代方案supersede。旧 API 直接以 Feature View 为操作对象需要你显式指定 Feature View 名称而push以数据源为中心Feast 自动推导出所有受影响的 Feature View语义更清晰、扩展性更好。旧接口仍保留用于兼容例如 feature_server.py 中/write-to-online-store端点仍调用它但新项目应优先使用 Push Source。是否需要 batch_sourcePush Source 可以可选地指定一个batch_source作为它的离线底表指定了 batch_source可以用于检索历史特征训练数据并且支持从离线存储物化到在线存储这一标准流程未指定 batch_source适用于特征只在训练之后生成、或者只在线使用例如向量数据库中的 embedding的场景。关于 batch_source 还有两点需要注意当你使用 batch_source 时你有责任保证数据也被写入该批量数据源如数据仓库即推送 写底表双写由你自己维护当 Push Source 被用作 Feature View 的流式源时Feature View 自身不需要再显式指定 batch_source——Feature View 会从 Push Source 继承 batch_source 信息。这一点也体现在序列化逻辑中PushSource._to_proto_impldata_source.py会把 batch_source 的timestamp_field、created_timestamp_column、field_mapping、date_partition_column继承到 Push Source 的 proto 中方便下游 Feature View 复用。典型流式数据架构流式数据是 Push Source 最重要的应用场景。一个典型的流式特征流水线如下原始事件到达流 1施加流式变换生成特征例如last_N_purchased_categories等聚合特征流 2可选把流 2 的值写入离线存储作为用于训练的历史日志把流 2 的值写入在线存储用于低延迟的特征服务周期性从离线存储物化特征到在线存储缩小训练与服务之间的偏差training-serving skew提升模型性能。Feast 同时支持两种推送方向把已在 Feature View 中注册的特征推送到在线存储以获得更新鲜的特征也可以指定推送到离线存储将一批流数据写入初始化 Feature Store 时仓库配置feature_store.yaml中声明的离线存储。基础示例定义 Push Source 并推送数据定义 Push Source下面的示例完整演示了如何定义一个以 BigQuery 表为底表的 Push Source并挂载到 Feature View 上。注意Push Source 的 schema 中必须包含实体列entity。from feast import Entity, PushSource, ValueType, BigQuerySource, FeatureView, Feature, Field from feast.types import Int64 push_source PushSource( namepush_source, batch_sourceBigQuerySource(tabletest.test), ) user Entity(nameuser, join_keys[user_id]) fv FeatureView( namefeature view, entities[user], schema[Field(namelife_time_value, dtypeInt64)], sourcepush_source, )对应到源码PushSource.__init__data_source.py只接收name、batch_source、description、tags、owner五个参数其中name是必填项batch_source为可选的DataSource类型。它继承自DataSource基类因此也具备timestamp_field、created_timestamp_column、field_mapping等通用数据源属性可从 batch_source 自动继承。推送数据推送时通过to参数控制目标存储它接受三种模式定义于 data_source.py 的PushMode枚举PushMode值行为PushMode.ONLINE1只写入在线存储默认值PushMode.OFFLINE2只写入离线存储PushMode.ONLINE_AND_OFFLINE3同时写入在线与离线存储to参数是可选的默认只推送到在线存储from feast import FeatureStore import pandas as pd from feast.data_source import PushMode fs FeatureStore(...) feature_data_frame pd.DataFrame() fs.push(push_source_name, feature_data_frame, toPushMode.ONLINE_AND_OFFLINE)FeatureStore.push的完整签名还包含allow_registry_cache是否允许使用注册表缓存默认 True与transform_on_write写入前是否执行变换默认 True两个参数见 feature_store.py。推送到离线存储时的列校验当to包含离线存储时会走write_to_offline_storefeature_store.py这里有一个容易踩坑的细节输入 DataFrame 的列必须与 Feature View 的 batch_source 列完全一致。源码会先取 batch_source 的列集合与输入 DataFrame 列集合做对比不一致时抛出 ValueError明确指出缺失列与多余列if input_columns_set ! source_columns_set: missing_expected_columns sorted(source_columns_set - input_columns_set) extra_unexpected_columns sorted(input_columns_set - source_columns_set) raise ValueError(...)reorder_columnsTrue默认时Feast 还会自动把 DataFrame 的列重排为与源表一致然后再通过provider.ingest_df_to_offline_store写入。Spark Streaming 集成示例在 PySpark 场景下最直接的做法是把 Feast Python SDK 引入已有的 PySpark 流水线在foreachBatch中把 Spark DataFrame 转为 pandas DataFrame 后推送from feast import FeatureStore store FeatureStore(...) spark SparkSession.builder.getOrCreate() streamingDF spark.readStream.format(...).load() def feast_writer(spark_df): pandas_df spark_df.to_pandas() store.push(driver_hourly_stats, pandas_df) streamingDF.writeStream.foreachBatch(feast_writer).start()这个模式可以被 contrib 流处理器在底层复用。仓库中与之对应的一组集成测试验证了端到端行为test_push_features_to_online_store.py验证推送到在线存储后能被get_online_features读取test_push_features_to_offline_store.py验证推送到离线存储后能用于历史检索test_stream_feature_view.py验证 StreamFeatureView 与 Push Source 的注册与解析。通过 Python Feature Server 推送在部署了 Python Feature Server 的场景下无需引入 SDK 即可通过 HTTP 接口推送数据。Feature Server 暴露了POST /push端点见 feature_server.py请求体为PushFeaturesRequest其中包含push_source_name、dfDataFrame 序列化数据、toonline/offline/online_and_offline、allow_registry_cache、transform_on_write等字段。to字段的解析映射与 SDK 的PushMode一一对应if request.to offline: to PushMode.OFFLINE elif request.to online: to PushMode.ONLINE elif request.to online_and_offline: to PushMode.ONLINE_AND_OFFLINE else: raise ValueError(...请指定 [online, offline, online_and_offline] 之一)服务端会先校验调用者的 RBAC 权限分别对应WRITE_ONLINE、WRITE_OFFLINE动作再根据 provider 是否支持异步在线写入决定走store.push_async还是线程池中的同步store.push。需要注意离线存储写入目前只有同步实现异步仅适用于在线存储部分见 feature_server.py 中的注释说明。此外若启用了离线推送批处理offline_push_batching_enabled见 feature_server.py离线推送会被OfflineWriteBatcher缓冲并按批写入此时 HTTP 响应码为 202 Accepted。完整部署与调用方式参见 Python Feature Server 文档。进阶StreamFeatureView 与构建流式特征Push Source 是构建StreamFeatureView的基础两者结合可以实现流式变换 → 推送 → 实时服务的完整链路。流式特征构建的完整教程包括示例项目、数据准备与端到端演示见 Tutorial: Building streaming features。在仓库的示例与测试仓库中也能看到典型的 Push Source StreamFeatureView 组合写法例如 feature_views.py 中的通用测试特征视图以及 example_feature_repo_1.py 中的示例特征仓库可作为定义 Feature View、Entity 与 Push Source 时的参考模板。小结Push Source 让特征值可以按需实时注入在线/离线存储取代旧的write_to_online_storeAPI一个 Push Source 可被多个 Feature View 共享一次推送自动传播到所有消费方batch_source可选有它可支持历史检索与物化无它适用于仅在线场景如 embedding推送方向由PushModeONLINE / OFFLINE / ONLINE_AND_OFFLINE控制默认仅在线生产环境可通过 Python Feature Server 的POST /push接口含 RBAC 权限校验与可选的离线批处理完成推送Spark Streaming 场景下在foreachBatch中调用store.push即可与现有 PySpark 流水线无缝集成。【免费下载链接】feastThe Open Source Feature Store for AI/ML项目地址: https://gitcode.com/GitHub_Trending/fe/feast创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考