Feast 流式窗口聚合的 Tiling 算法与中间表示IR机制解析【免费下载链接】feastThe Open Source Feature Store for AI/ML项目地址: https://gitcode.com/GitHub_Trending/fe/feast本文围绕 Feast 开源特征存储中feast.aggregation.tiling模块模块文档展开深入讲解其面向流式时间窗口聚合优化的 tiling分片算法如何在毫秒级时间片上预聚合数据、如何借助中间表示Intermediate RepresentationsIR实现所有聚合类型在 tile 合并时的数学正确性以及 Spark/Ray 计算引擎如何接入这套纯 pandas 的核心逻辑。读完本文你将理解StreamFeatureView中enable_tiling/tiling_hop_size参数背后的实现原理并掌握 tiling 模块的完整调用链与配置方法。模块定位为什么需要 TilingTiling 是 Feast 针对流式时间窗口聚合如来自 Kafka、Kinesis、PushSource 的高频更新特征的优化技术。其核心思路是把数据按固定大小的 hop 间隔预先聚合成瓦片tile在多次窗口刷新之间复用已计算的 tile而不是每次都从原始事件全量重算。从模块源码的架构注释sdk/python/feast/aggregation/tiling/__init__.py可以看到tiling 的整体数据流分四步引擎节点把引擎数据转换为 pandas如dataset.to_pandas()、toPandas()orchestrator.py生成累积 tilecumulative tilestile_subtraction.py把累积 tile 转换为窗口聚合windowed aggregations引擎节点把 pandas 结果转回引擎格式如from_pandas()、createDataFrame()。该设计保证了核心算法与具体计算引擎解耦核心逻辑是纯 pandas 实现任何引擎Spark、Ray 等只需做 DataFrame 的来回转换。为什么需要中间表示IR聚合的合并难题传统做法有两条路各有利弊每次从原始数据重算——正确但慢、代价高每个 tile 只存最终聚合值——快但合并时常常不正确。以平均值avg为例WRONG: avg(tile1, tile2) ≠ (avg_tile1 avg_tile2) / 2 tile1: [10, 20, 30] → avg 20 tile2: [100] → avg 100 正确合并结果: (102030100) / 4 40 错误合并结果: (20 100) / 2 60同样的问题也存在于std标准差、var方差等需要看到全部数值才能计算的holistic整体性聚合。解决方案就是不存最终值改存可正确合并的中间数据IR例如avg存sum与count合并时先把各自的和与计数相加再相除。核心数据结构Aggregation 与 IRMetadatafeast.aggregation.tiling模块对外暴露五个成员见 模块__all__IRMetadata、get_ir_metadata_for_aggregation、apply_sawtooth_window_tiling、convert_cumulative_to_windowed、deduplicate_keep_latest。Aggregation聚合规格聚合规格由feast/aggregation/__init__.py中的Aggregation类定义核心字段column被聚合的源列名function内置聚合函数支持sum、max、min、count、mean、count_distincttime_window聚合时间窗口timedeltaslide_interval滑动/跳跃间隔未指定时默认等于time_windowname可选输出特征名默认推导为{function}_{column}指定了time_window时追加_{seconds}s后缀见resolved_name如sum_amount_3600s。该类的to_proto/from_proto方法负责与Aggregation.proto互转time_window与slide_interval序列化为 protobufDuration。IRMetadata聚合的中间表示元信息定义于sdk/python/feast/aggregation/tiling/base.pydataclass class IRMetadata: type: str # algebraic 或 holistic ir_columns: Optional[List[str]] None # IR 列名列表 computation: Optional[str] None # 由 IR 计算最终值的公式描述typealgebraic代数聚合sum、count、max、min不需要 IR 列tile 的最终值本身就是可直接合并的值typeholistic或avg整体聚合avg、std、var必须存储额外 IR 列才能正确合并。get_ir_metadata_for_aggregation为聚合选择正确的 IR 存储方案该函数base.py第 30-97 行根据聚合类型决定需要存哪些中间值聚合函数IR 类型需要存储的 IR 列最终值计算sum/count/max/minalgebraic无自身即 IR直接使用avg/meanavg_tail_{name}_sum、_tail_{name}_countsum / countstd/stddev/var/varianceholistic_tail_{name}_sum、_tail_{name}_count、_tail_{name}_sum_sqsqrt((sum_sq - sum²/count) / (count-1))count_distinct—不支持 tiling直接抛出ValueError需设置enable_tilingFalse或改用代数聚合对于count_distinct源码给出了明确报错信息count_distinct does not support tiling. Use enable_tilingFalse or choose an algebraic aggregation (sum, count, min, max).这也从实现层面说明tiling 并不覆盖全部聚合类型count_distinct这类依赖全局去重的聚合无法用 tile 合并的方式正确计算。未知聚合类型会被保守地当作 algebraic 处理。apply_sawtooth_window_tiling累积 tile 的生成orchestrator.py中的apply_sawtooth_window_tiling是 tiling 的核心入口输入 pandas DataFrame、聚合列表、分组键、时间戳列、窗口大小与 hop 大小输出带累积尾部聚合的 tile。其算法分五步计算 hop 间隔列_hop_interval先把时间戳统一转成毫秒整数datetime64 类型直接astype(int64) // 10**6否则先pd.to_datetime再按timestamp_ms // hop_size_ms * hop_size_ms计算每个事件所属 hop 的下边界inclusive lower boundary按实体键 hop 间隔分组聚合根据 IR 元信息构造 pandas 聚合字典。代数聚合直接sum/count/max/minavg聚合出 sum 与 countstd/var额外用lambda x: (x**2).sum()计算平方和计算累积和为每个实体生成从最小 hop 到最大 hop 的完整 hop 网格保证没有数据的区间也有 tile左连接实际聚合结果、缺失 IR 填 0再对各 IR 列做cumsum()从而把 hop 内聚合转换为从窗口起点到当前 T的累积 tile附加 tile 元数据_tile_start _hop_interval_tile_end _hop_interval hop_size_ms由 IR 计算最终特征值代数聚合直接取 IR 列avg用np.where(count 0, sum/count, 0)计算。sawtooth锯齿窗命名的含义即在此tile 在时间 T 处包含从窗口起点到 T 的聚合后续只需做减法即可还原任意滑动窗口。convert_cumulative_to_windowed累积 tile 到窗口聚合的转换tile_subtraction.py中的convert_cumulative_to_windowed完成核心数学运算windowed_agg_at_T cumulative_tile_at_T - cumulative_tile_at_(T-window_size)实现要点按实体键分组按_tile_end排序保证时序正确对每个 tile计算window_start tile_end - window_size_ms只使用精确匹配的_tile_end window_start的先前 tile 做减法——源码注释明确指出若无精确匹配任何更早的 tile 都会导致计算出的窗口大于请求窗口因此视为无前驱 tile用 0 代替代数聚合sum/count直接做current - prevmax/min因为无法通过减法还原语义上直接取当前累积值holistic 聚合先对每个 IR 分量做减法windowed_IR current_IR - previous_IR再由窗口化后的 IR 重新计算最终值如avg windowed_sum / windowed_countcount 0时置 0输出行的event_timestamp设为 tile 结束时间pd.to_datetime(tile_end, unitms)最后删除_tile_start、_tile_end、_hop_interval等内部列返回可直接写入在线存储的窗口化结果。deduplicate_keep_latest每个实体保留最新时间戳deduplicate_keep_latest是一个简洁的收尾工具按时间戳降序排序后groupby(entity_keys).first()为每个实体只保留最新一条记录。它在引擎节点中用于把多行 tile 结果收敛为每个实体一行的最终输出。配置入口StreamFeatureView 的 tiling 参数tiling 由StreamFeatureView的两个参数控制enable_tiling默认False是否启用 tiling 优化流式场景建议开启tiling_hop_size默认timedelta(minutes5)tile 的时间间隔即流式更新频率。配置示例结合 tiling 概念文档from feast import StreamFeatureView, Aggregation from feast.data_source import PushSource, KafkaSource from datetime import timedelta customer_features StreamFeatureView( namecustomer_transaction_features, entities[customer], sourceKafkaSource( nametransactions_stream, kafka_bootstrap_serverslocalhost:9092, topictransactions, timestamp_fieldevent_timestamp, batch_sourcefile_source, # 历史数据回填 ), aggregations[ Aggregation(columnamount, functionsum, time_windowtimedelta(hours1), namesum_amount_1h), Aggregation(columnamount, functionavg, time_windowtimedelta(hours1), nameavg_amount_1h), Aggregation(columnamount, functionstd, time_windowtimedelta(hours1), namestd_amount_1h), ], timestamp_fieldevent_timestamp, onlineTrue, # Tiling 配置 enable_tilingTrue, # 流式场景提速 tiling_hop_sizetimedelta(minutes5), # 更新频率 )源码中的关键约束stream_feature_view.py第 167-181 行指定了aggregations时必须提供timestamp_field否则直接抛ValueErroraggregations must have a timestamp field associated with them to perform the aggregationsenable_tilingTrue时有效 hop 大小必须严格小于所有聚合中最小的时间窗口否则抛ValueErrortiling_hop_size must be smaller than the minimum aggregation time_window ... the tiling algorithm will produce incorrect results——这是保证锯齿窗减法边界正确的数学前提tiling_hop_size通过to_proto/from_proto序列化为 protobufDuration参与StreamFeatureViewSpec的注册与反序列化相关字段定义见feast/core/StreamFeatureView.proto的 stub 声明 StreamFeatureView_pb2.pyi。关于 hop 大小的取舍官方概念文档给出的指引是更小的 hop 更细粒度的 tile处理窗口期内内存占用可能更高更大的 hop 更粗粒度内存占用可能更低。计算引擎接入Spark 与 Ray 节点tiling 的引擎接入点位于两处节点实现中Sparksdk/python/feast/infra/compute_engines/spark/nodes.py第 267-346 行附近Raysdk/python/feast/infra/compute_engines/ray/nodes.py第 364-442 行附近。两者的调用模式完全一致均遵循模块架构注释描述的四步流水线引擎侧数据.toPandas()/.to_pandas()转成 pandas调用apply_sawtooth_window_tiling生成累积 tile调用convert_cumulative_to_windowed转为窗口化聚合调用deduplicate_keep_latest收敛每实体一行再createDataFrame()/from_pandas()转回引擎格式。引擎节点在启用 tiling 时还有一个前置校验两个nodes.py中逻辑一致if self.enable_tiling and has_time_windows:且要求所有聚合都设置time_window否则报错提示 Either set time_window for all aggregations or disable tiling by setting enable_tilingFalse见 Spark 节点第 244-263 行。而feature_builder.py则负责从StreamFeatureView对象读取 tiling 配置并传给节点spark/feature_builder.py第 58-68 行、ray/feature_builder.py第 80-90 行形成视图配置 → builder → 节点 → tiling 算法的完整调用链。适用场景与边界根据 Feast 官方 tiling 概念文档docs/getting-started/concepts/tiling.mdtiling 的主要收益场景是流式高频更新例如每小时窗口、每 5 分钟刷新的场景下每次更新只需计算 1 个新 tile、复用内存中的 11 个旧 tile即约 92% 的 tile 复用率而非每次扫描整小时窗口的上千条事件。边界条件同样明确Local Compute Engine 不支持时间窗口聚合也就无法使用 tilingSpark 与 Ray 计算引擎对流式与批处理均完整支持 tilingcount_distinct不支持 tiling需关闭该功能或改用代数聚合保证正确性的硬性前提是tiling_hop_size min(time_window)且窗口边界依赖 tile 端点的精确匹配。综上feast.aggregation.tiling模块以纯 pandas 核心算法 引擎侧薄转换层的架构为 Feast 流式时间窗口聚合提供了可增量更新、可跨引擎复用的高效实现IR 机制则保证了avg/std/var等整体性聚合在 tile 合并过程中的数学正确性。理解这五个公开 API 与StreamFeatureView的参数约束即可在生产流式特征管线中正确启用并调优 tiling。【免费下载链接】feastThe Open Source Feature Store for AI/ML项目地址: https://gitcode.com/GitHub_Trending/fe/feast创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Feast 流式窗口聚合的 Tiling 算法与中间表示（IR）机制解析
Feast 流式窗口聚合的 Tiling 算法与中间表示IR机制解析【免费下载链接】feastThe Open Source Feature Store for AI/ML项目地址: https://gitcode.com/GitHub_Trending/fe/feast本文围绕 Feast 开源特征存储中feast.aggregation.tiling模块模块文档展开深入讲解其面向流式时间窗口聚合优化的 tiling分片算法如何在毫秒级时间片上预聚合数据、如何借助中间表示Intermediate RepresentationsIR实现所有聚合类型在 tile 合并时的数学正确性以及 Spark/Ray 计算引擎如何接入这套纯 pandas 的核心逻辑。读完本文你将理解StreamFeatureView中enable_tiling/tiling_hop_size参数背后的实现原理并掌握 tiling 模块的完整调用链与配置方法。模块定位为什么需要 TilingTiling 是 Feast 针对流式时间窗口聚合如来自 Kafka、Kinesis、PushSource 的高频更新特征的优化技术。其核心思路是把数据按固定大小的 hop 间隔预先聚合成瓦片tile在多次窗口刷新之间复用已计算的 tile而不是每次都从原始事件全量重算。从模块源码的架构注释sdk/python/feast/aggregation/tiling/__init__.py可以看到tiling 的整体数据流分四步引擎节点把引擎数据转换为 pandas如dataset.to_pandas()、toPandas()orchestrator.py生成累积 tilecumulative tilestile_subtraction.py把累积 tile 转换为窗口聚合windowed aggregations引擎节点把 pandas 结果转回引擎格式如from_pandas()、createDataFrame()。该设计保证了核心算法与具体计算引擎解耦核心逻辑是纯 pandas 实现任何引擎Spark、Ray 等只需做 DataFrame 的来回转换。为什么需要中间表示IR聚合的合并难题传统做法有两条路各有利弊每次从原始数据重算——正确但慢、代价高每个 tile 只存最终聚合值——快但合并时常常不正确。以平均值avg为例WRONG: avg(tile1, tile2) ≠ (avg_tile1 avg_tile2) / 2 tile1: [10, 20, 30] → avg 20 tile2: [100] → avg 100 正确合并结果: (102030100) / 4 40 错误合并结果: (20 100) / 2 60同样的问题也存在于std标准差、var方差等需要看到全部数值才能计算的holistic整体性聚合。解决方案就是不存最终值改存可正确合并的中间数据IR例如avg存sum与count合并时先把各自的和与计数相加再相除。核心数据结构Aggregation 与 IRMetadatafeast.aggregation.tiling模块对外暴露五个成员见 模块__all__IRMetadata、get_ir_metadata_for_aggregation、apply_sawtooth_window_tiling、convert_cumulative_to_windowed、deduplicate_keep_latest。Aggregation聚合规格聚合规格由feast/aggregation/__init__.py中的Aggregation类定义核心字段column被聚合的源列名function内置聚合函数支持sum、max、min、count、mean、count_distincttime_window聚合时间窗口timedeltaslide_interval滑动/跳跃间隔未指定时默认等于time_windowname可选输出特征名默认推导为{function}_{column}指定了time_window时追加_{seconds}s后缀见resolved_name如sum_amount_3600s。该类的to_proto/from_proto方法负责与Aggregation.proto互转time_window与slide_interval序列化为 protobufDuration。IRMetadata聚合的中间表示元信息定义于sdk/python/feast/aggregation/tiling/base.pydataclass class IRMetadata: type: str # algebraic 或 holistic ir_columns: Optional[List[str]] None # IR 列名列表 computation: Optional[str] None # 由 IR 计算最终值的公式描述typealgebraic代数聚合sum、count、max、min不需要 IR 列tile 的最终值本身就是可直接合并的值typeholistic或avg整体聚合avg、std、var必须存储额外 IR 列才能正确合并。get_ir_metadata_for_aggregation为聚合选择正确的 IR 存储方案该函数base.py第 30-97 行根据聚合类型决定需要存哪些中间值聚合函数IR 类型需要存储的 IR 列最终值计算sum/count/max/minalgebraic无自身即 IR直接使用avg/meanavg_tail_{name}_sum、_tail_{name}_countsum / countstd/stddev/var/varianceholistic_tail_{name}_sum、_tail_{name}_count、_tail_{name}_sum_sqsqrt((sum_sq - sum²/count) / (count-1))count_distinct—不支持 tiling直接抛出ValueError需设置enable_tilingFalse或改用代数聚合对于count_distinct源码给出了明确报错信息count_distinct does not support tiling. Use enable_tilingFalse or choose an algebraic aggregation (sum, count, min, max).这也从实现层面说明tiling 并不覆盖全部聚合类型count_distinct这类依赖全局去重的聚合无法用 tile 合并的方式正确计算。未知聚合类型会被保守地当作 algebraic 处理。apply_sawtooth_window_tiling累积 tile 的生成orchestrator.py中的apply_sawtooth_window_tiling是 tiling 的核心入口输入 pandas DataFrame、聚合列表、分组键、时间戳列、窗口大小与 hop 大小输出带累积尾部聚合的 tile。其算法分五步计算 hop 间隔列_hop_interval先把时间戳统一转成毫秒整数datetime64 类型直接astype(int64) // 10**6否则先pd.to_datetime再按timestamp_ms // hop_size_ms * hop_size_ms计算每个事件所属 hop 的下边界inclusive lower boundary按实体键 hop 间隔分组聚合根据 IR 元信息构造 pandas 聚合字典。代数聚合直接sum/count/max/minavg聚合出 sum 与 countstd/var额外用lambda x: (x**2).sum()计算平方和计算累积和为每个实体生成从最小 hop 到最大 hop 的完整 hop 网格保证没有数据的区间也有 tile左连接实际聚合结果、缺失 IR 填 0再对各 IR 列做cumsum()从而把 hop 内聚合转换为从窗口起点到当前 T的累积 tile附加 tile 元数据_tile_start _hop_interval_tile_end _hop_interval hop_size_ms由 IR 计算最终特征值代数聚合直接取 IR 列avg用np.where(count 0, sum/count, 0)计算。sawtooth锯齿窗命名的含义即在此tile 在时间 T 处包含从窗口起点到 T 的聚合后续只需做减法即可还原任意滑动窗口。convert_cumulative_to_windowed累积 tile 到窗口聚合的转换tile_subtraction.py中的convert_cumulative_to_windowed完成核心数学运算windowed_agg_at_T cumulative_tile_at_T - cumulative_tile_at_(T-window_size)实现要点按实体键分组按_tile_end排序保证时序正确对每个 tile计算window_start tile_end - window_size_ms只使用精确匹配的_tile_end window_start的先前 tile 做减法——源码注释明确指出若无精确匹配任何更早的 tile 都会导致计算出的窗口大于请求窗口因此视为无前驱 tile用 0 代替代数聚合sum/count直接做current - prevmax/min因为无法通过减法还原语义上直接取当前累积值holistic 聚合先对每个 IR 分量做减法windowed_IR current_IR - previous_IR再由窗口化后的 IR 重新计算最终值如avg windowed_sum / windowed_countcount 0时置 0输出行的event_timestamp设为 tile 结束时间pd.to_datetime(tile_end, unitms)最后删除_tile_start、_tile_end、_hop_interval等内部列返回可直接写入在线存储的窗口化结果。deduplicate_keep_latest每个实体保留最新时间戳deduplicate_keep_latest是一个简洁的收尾工具按时间戳降序排序后groupby(entity_keys).first()为每个实体只保留最新一条记录。它在引擎节点中用于把多行 tile 结果收敛为每个实体一行的最终输出。配置入口StreamFeatureView 的 tiling 参数tiling 由StreamFeatureView的两个参数控制enable_tiling默认False是否启用 tiling 优化流式场景建议开启tiling_hop_size默认timedelta(minutes5)tile 的时间间隔即流式更新频率。配置示例结合 tiling 概念文档from feast import StreamFeatureView, Aggregation from feast.data_source import PushSource, KafkaSource from datetime import timedelta customer_features StreamFeatureView( namecustomer_transaction_features, entities[customer], sourceKafkaSource( nametransactions_stream, kafka_bootstrap_serverslocalhost:9092, topictransactions, timestamp_fieldevent_timestamp, batch_sourcefile_source, # 历史数据回填 ), aggregations[ Aggregation(columnamount, functionsum, time_windowtimedelta(hours1), namesum_amount_1h), Aggregation(columnamount, functionavg, time_windowtimedelta(hours1), nameavg_amount_1h), Aggregation(columnamount, functionstd, time_windowtimedelta(hours1), namestd_amount_1h), ], timestamp_fieldevent_timestamp, onlineTrue, # Tiling 配置 enable_tilingTrue, # 流式场景提速 tiling_hop_sizetimedelta(minutes5), # 更新频率 )源码中的关键约束stream_feature_view.py第 167-181 行指定了aggregations时必须提供timestamp_field否则直接抛ValueErroraggregations must have a timestamp field associated with them to perform the aggregationsenable_tilingTrue时有效 hop 大小必须严格小于所有聚合中最小的时间窗口否则抛ValueErrortiling_hop_size must be smaller than the minimum aggregation time_window ... the tiling algorithm will produce incorrect results——这是保证锯齿窗减法边界正确的数学前提tiling_hop_size通过to_proto/from_proto序列化为 protobufDuration参与StreamFeatureViewSpec的注册与反序列化相关字段定义见feast/core/StreamFeatureView.proto的 stub 声明 StreamFeatureView_pb2.pyi。关于 hop 大小的取舍官方概念文档给出的指引是更小的 hop 更细粒度的 tile处理窗口期内内存占用可能更高更大的 hop 更粗粒度内存占用可能更低。计算引擎接入Spark 与 Ray 节点tiling 的引擎接入点位于两处节点实现中Sparksdk/python/feast/infra/compute_engines/spark/nodes.py第 267-346 行附近Raysdk/python/feast/infra/compute_engines/ray/nodes.py第 364-442 行附近。两者的调用模式完全一致均遵循模块架构注释描述的四步流水线引擎侧数据.toPandas()/.to_pandas()转成 pandas调用apply_sawtooth_window_tiling生成累积 tile调用convert_cumulative_to_windowed转为窗口化聚合调用deduplicate_keep_latest收敛每实体一行再createDataFrame()/from_pandas()转回引擎格式。引擎节点在启用 tiling 时还有一个前置校验两个nodes.py中逻辑一致if self.enable_tiling and has_time_windows:且要求所有聚合都设置time_window否则报错提示 Either set time_window for all aggregations or disable tiling by setting enable_tilingFalse见 Spark 节点第 244-263 行。而feature_builder.py则负责从StreamFeatureView对象读取 tiling 配置并传给节点spark/feature_builder.py第 58-68 行、ray/feature_builder.py第 80-90 行形成视图配置 → builder → 节点 → tiling 算法的完整调用链。适用场景与边界根据 Feast 官方 tiling 概念文档docs/getting-started/concepts/tiling.mdtiling 的主要收益场景是流式高频更新例如每小时窗口、每 5 分钟刷新的场景下每次更新只需计算 1 个新 tile、复用内存中的 11 个旧 tile即约 92% 的 tile 复用率而非每次扫描整小时窗口的上千条事件。边界条件同样明确Local Compute Engine 不支持时间窗口聚合也就无法使用 tilingSpark 与 Ray 计算引擎对流式与批处理均完整支持 tilingcount_distinct不支持 tiling需关闭该功能或改用代数聚合保证正确性的硬性前提是tiling_hop_size min(time_window)且窗口边界依赖 tile 端点的精确匹配。综上feast.aggregation.tiling模块以纯 pandas 核心算法 引擎侧薄转换层的架构为 Feast 流式时间窗口聚合提供了可增量更新、可跨引擎复用的高效实现IR 机制则保证了avg/std/var等整体性聚合在 tile 合并过程中的数学正确性。理解这五个公开 API 与StreamFeatureView的参数约束即可在生产流式特征管线中正确启用并调优 tiling。【免费下载链接】feastThe Open Source Feature Store for AI/ML项目地址: https://gitcode.com/GitHub_Trending/fe/feast创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
