Ray node for performing aggregations on feature data.
| 316 | |
| 317 | |
| 318 | class RayAggregationNode(DAGNode): |
| 319 | """ |
| 320 | Ray node for performing aggregations on feature data. |
| 321 | """ |
| 322 | |
| 323 | def __init__( |
| 324 | self, |
| 325 | name: str, |
| 326 | aggregations: List[Aggregation], |
| 327 | group_by_keys: List[str], |
| 328 | timestamp_col: str, |
| 329 | config: RayComputeEngineConfig, |
| 330 | enable_tiling: bool = False, |
| 331 | hop_size: Optional[timedelta] = None, |
| 332 | ): |
| 333 | super().__init__(name) |
| 334 | self.aggregations = aggregations |
| 335 | self.group_by_keys = group_by_keys |
| 336 | self.timestamp_col = timestamp_col |
| 337 | self.config = config |
| 338 | self.enable_tiling = enable_tiling |
| 339 | self.hop_size = hop_size |
| 340 | |
| 341 | def execute(self, context: ExecutionContext) -> DAGValue: |
| 342 | """Execute the aggregation operation.""" |
| 343 | input_value = self.get_single_input_value(context) |
| 344 | input_value.assert_format(DAGFormat.RAY) |
| 345 | dataset: Dataset = input_value.data |
| 346 | |
| 347 | # Check if tiling should be used |
| 348 | has_time_windows = any(agg.time_window for agg in self.aggregations) |
| 349 | if self.enable_tiling and has_time_windows: |
| 350 | return self._execute_tiled_aggregation(dataset) |
| 351 | else: |
| 352 | return self._execute_standard_aggregation(dataset) |
| 353 | |
| 354 | def _execute_tiled_aggregation(self, dataset: Dataset) -> DAGValue: |
| 355 | """ |
| 356 | Execute tiled aggregation. |
| 357 | |
| 358 | Flow: |
| 359 | 1. Convert Ray Dataset → pandas |
| 360 | 2. Generate cumulative tiles |
| 361 | 3. Convert to windowed aggregations |
| 362 | 4. Convert pandas → Ray Dataset |
| 363 | """ |
| 364 | from feast.aggregation.tiling.orchestrator import apply_sawtooth_window_tiling |
| 365 | from feast.aggregation.tiling.tile_subtraction import ( |
| 366 | convert_cumulative_to_windowed, |
| 367 | deduplicate_keep_latest, |
| 368 | ) |
| 369 | |
| 370 | ray_wrapper = get_ray_wrapper() |
| 371 | |
| 372 | input_pdf = dataset.to_pandas() |
| 373 | |
| 374 | for agg in self.aggregations: |
| 375 | if agg.time_window is None: |
no outgoing calls