Skip to content

Commit af613d4

Browse files
ntkatholentkathole
committed
fix: Handle comments and removed use_ray_cluster config
Signed-off-by: ntkathole <nikhilkathole2683@gmail.com>
1 parent c3a87ab commit af613d4

14 files changed

Lines changed: 87 additions & 107 deletions

File tree

‎docs/reference/compute-engine/ray.md‎

Lines changed: 17 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -42,29 +42,31 @@ offline_store:
4242
storage_path: data/ray_storage
4343
batch_engine:
4444
type: ray.engine
45-
max_workers: 4 # Optional: Maximum number of workers
46-
enable_optimization: true # Optional: Enable performance optimizations
45+
max_workers: 4 # Optional: Maximum number of workers
46+
enable_optimization: true # Optional: Enable performance optimizations
4747
broadcast_join_threshold_mb: 100 # Optional: Broadcast join threshold (MB)
4848
max_parallelism_multiplier: 2 # Optional: Parallelism multiplier
4949
target_partition_size_mb: 64 # Optional: Target partition size (MB)
5050
window_size_for_joins: "1H" # Optional: Time window for distributed joins
5151
ray_address: localhost:10001 # Optional: Ray cluster address
52-
use_ray_cluster: false # Optional: Use Ray cluster mode
5352
```
5453
5554
### Configuration Options
5655
5756
| Option | Type | Default | Description |
5857
|--------|------|---------|-------------|
59-
| `type` | string | Required | Must be `ray.engine` |
60-
| `max_workers` | int | CPU count | Maximum number of Ray workers |
58+
| `type` | string | `"ray.engine"` | Must be `ray.engine` |
59+
| `max_workers` | int | None (uses all cores) | Maximum number of Ray workers |
6160
| `enable_optimization` | boolean | true | Enable performance optimizations |
6261
| `broadcast_join_threshold_mb` | int | 100 | Size threshold for broadcast joins (MB) |
6362
| `max_parallelism_multiplier` | int | 2 | Parallelism as multiple of CPU cores |
6463
| `target_partition_size_mb` | int | 64 | Target partition size (MB) |
6564
| `window_size_for_joins` | string | "1H" | Time window for distributed joins |
66-
| `ray_address` | string | None | Ray cluster address |
67-
| `use_ray_cluster` | boolean | false | Use Ray cluster mode |
65+
| `ray_address` | string | None | Ray cluster address (None = local Ray) |
66+
| `enable_distributed_joins` | boolean | true | Enable distributed joins for large datasets |
67+
| `staging_location` | string | None | Remote path for batch materialization jobs |
68+
| `ray_conf` | dict | None | Ray configuration parameters |
69+
| `execution_timeout_seconds` | int | None | Timeout for job execution in seconds |
6870

6971
## Usage Examples
7072

@@ -159,7 +161,6 @@ batch_engine:
159161
window_size_for_joins: "30min"
160162
161163
# Ray cluster configuration
162-
use_ray_cluster: true
163164
ray_address: "ray://head-node:10001"
164165
```
165166

@@ -181,7 +182,6 @@ offline_store:
181182
type: ray
182183
storage_path: s3://my-bucket/feast-data # Optional: Path for storing datasets
183184
ray_address: localhost:10001 # Optional: Ray cluster address
184-
use_ray_cluster: true # Optional: Use Ray cluster mode
185185
186186
# Ray compute engine configuration
187187
# Handles complex feature computation and distributed processing
@@ -202,7 +202,6 @@ batch_engine:
202202
203203
# Ray cluster configuration (inherits from offline_store if not specified)
204204
ray_address: localhost:10001 # Ray cluster address
205-
use_ray_cluster: true # Use Ray cluster mode
206205
```
207206

208207
## DAG Node Types
@@ -258,22 +257,18 @@ The Ray compute engine automatically selects optimal join strategies:
258257
### Broadcast Join
259258

260259
Used for small feature datasets:
261-
```python
262-
# Automatically selected when feature data < 100MB
263-
# Features are cached in Ray's object store
264-
# Entities are distributed across cluster
265-
# Each worker gets a copy of feature data
266-
```
260+
- Automatically selected when feature data < 100MB
261+
- Features are cached in Ray's object store
262+
- Entities are distributed across cluster
263+
- Each worker gets a copy of feature data
267264

268265
### Distributed Windowed Join
269266

270267
Used for large feature datasets:
271-
```python
272-
# Automatically selected when feature data > 100MB
273-
# Data is partitioned by time windows
274-
# Point-in-time joins within each window
275-
# Results are combined across windows
276-
```
268+
- Automatically selected when feature data > 100MB
269+
- Data is partitioned by time windows
270+
- Point-in-time joins within each window
271+
- Results are combined across windows
277272

278273
### Strategy Selection Logic
279274

@@ -353,7 +348,6 @@ offline_store:
353348
storage_path: s3://my-bucket/feast-data
354349
batch_engine:
355350
type: ray.engine
356-
use_ray_cluster: true
357351
ray_address: "ray://ray-cluster:10001"
358352
broadcast_join_threshold_mb: 50
359353
```

‎docs/reference/offline-stores/ray.md‎

Lines changed: 0 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,6 @@ The Ray offline store provides:
3333
| export to arrow table | Yes |
3434
| persist results in offline store| Yes |
3535
| local execution of ODFVs | Yes |
36-
| remote execution of ODFVs | No |
3736
| preview query plan | Yes |
3837
| read partitioned data | Yes |
3938

@@ -74,7 +73,6 @@ offline_store:
7473
type: ray
7574
storage_path: data/ray_storage # Optional: Path for storing datasets
7675
ray_address: localhost:10001 # Optional: Ray cluster address
77-
use_ray_cluster: false # Optional: Whether to use Ray cluster
7876
```
7977
8078
### Ray Offline Store + Compute Engine
@@ -91,7 +89,6 @@ offline_store:
9189
type: ray
9290
storage_path: s3://my-bucket/feast-data # Optional: Path for storing datasets
9391
ray_address: localhost:10001 # Optional: Ray cluster address
94-
use_ray_cluster: true # Optional: Use Ray cluster mode
9592

9693
# Ray compute engine for distributed feature processing
9794
batch_engine:
@@ -112,7 +109,6 @@ batch_engine:
112109

113110
# Ray cluster configuration (optional)
114111
ray_address: localhost:10001 # Ray cluster address
115-
use_ray_cluster: true # Use Ray cluster mode
116112
staging_location: s3://my-bucket/staging # Remote staging location
117113
```
118114
@@ -158,7 +154,6 @@ offline_store:
158154
type: ray
159155
storage_path: s3://my-production-bucket/feast-data
160156
ray_address: "ray://production-head-node:10001"
161-
use_ray_cluster: true
162157

163158
batch_engine:
164159
type: ray.engine
@@ -169,7 +164,6 @@ batch_engine:
169164
target_partition_size_mb: 128
170165
window_size_for_joins: "30min"
171166
ray_address: "ray://production-head-node:10001"
172-
use_ray_cluster: true
173167
staging_location: s3://my-production-bucket/staging
174168
```
175169
@@ -182,7 +176,6 @@ batch_engine:
182176
| `type` | string | Required | Must be `feast.offline_stores.contrib.ray_offline_store.ray.RayOfflineStore` or `ray` |
183177
| `storage_path` | string | None | Path for storing temporary files and datasets |
184178
| `ray_address` | string | None | Address of the Ray cluster (e.g., "localhost:10001") |
185-
| `use_ray_cluster` | boolean | false | Whether to use Ray cluster mode |
186179
| `ray_conf` | dict | None | Ray initialization parameters for resource management (e.g., memory, CPU limits) |
187180

188181
#### Ray Compute Engine Options
@@ -224,7 +217,6 @@ offline_store:
224217
type: ray
225218
storage_path: s3://my-bucket/feast-data
226219
ray_address: "ray://production-cluster:10001"
227-
use_ray_cluster: true
228220
# Optimized for production workloads
229221
broadcast_join_threshold_mb: 100
230222
max_parallelism_multiplier: 2
@@ -259,7 +251,6 @@ offline_store:
259251
offline_store:
260252
type: ray
261253
ray_address: "ray://cluster-head:10001"
262-
use_ray_cluster: true
263254
broadcast_join_threshold_mb: 200
264255
max_parallelism_multiplier: 4
265256
```
@@ -406,7 +397,6 @@ ray start --head --port=10001
406397
offline_store:
407398
type: ray
408399
ray_address: localhost:10001
409-
use_ray_cluster: true
410400
storage_path: s3://my-bucket/features
411401
```
412402

@@ -468,7 +458,6 @@ offline_store:
468458
offline_store:
469459
type: ray
470460
storage_path: s3://my-bucket/feast-data
471-
use_ray_cluster: true
472461
473462
batch_engine:
474463
type: ray.engine

‎sdk/python/feast/infra/compute_engines/ray/compute.py‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,7 @@ def __init__(
5858
def _ensure_ray_initialized(self):
5959
"""Ensure Ray is initialized with proper configuration."""
6060
if not ray.is_initialized():
61-
if self.config.use_ray_cluster and self.config.ray_address:
61+
if self.config.ray_address:
6262
ray.init(
6363
address=self.config.ray_address,
6464
ignore_reinit_error=True,
@@ -206,7 +206,7 @@ def _materialize_from_offline_store(
206206
if getattr(feature_view, "online", False):
207207
# TODO: Implement proper online store writing with correct data format conversion
208208
logger.debug(
209-
f"Online store writing not implemented yet for {arrow_table.num_rows} rows"
209+
"Online store writing not implemented yet for Ray compute engine"
210210
)
211211

212212
# Write to offline store if enabled (this handles sink_source automatically for derived views)
@@ -228,9 +228,7 @@ def _materialize_from_offline_store(
228228

229229
# Write to sink_source using Ray data
230230
try:
231-
# Convert arrow table to pandas then to ray dataset
232-
df = arrow_table.to_pandas()
233-
ray_dataset = ray.data.from_pandas(df)
231+
ray_dataset = ray.data.from_arrow(arrow_table)
234232
ray_dataset.write_parquet(sink_source.path)
235233
except Exception as e:
236234
logger.error(

‎sdk/python/feast/infra/compute_engines/ray/config.py‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,6 @@ class RayComputeEngineConfig(FeastConfigBaseModel):
1717
ray_address: Optional[str] = None
1818
"""Ray cluster address. If None, uses local Ray cluster."""
1919

20-
use_ray_cluster: bool = False
21-
"""Whether to use an existing Ray cluster."""
22-
2320
staging_location: Optional[StrictStr] = None
2421
"""Remote path for batch materialization jobs"""
2522

‎sdk/python/feast/infra/compute_engines/ray/feature_builder.py‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -169,6 +169,9 @@ def build_output_nodes(self, view, final_node):
169169
def build_validation_node(self, view, input_node):
170170
"""Build the validation node for feature validation."""
171171
# TODO: Implement validation logic
172+
logger.warning(
173+
"Feature validation is not yet implemented for Ray compute engine."
174+
)
172175
return input_node
173176

174177
def _build(self, view, input_nodes: Optional[List[DAGNode]]) -> DAGNode:

‎sdk/python/feast/infra/compute_engines/ray/job.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -184,6 +184,7 @@ def to_remote_storage(self) -> list[str]:
184184
f"{self._config.batch_engine.staging_location}/{str(uuid.uuid4())}"
185185
)
186186
self._result_dataset.write_parquet(output_uri)
187+
logger.debug(f"Wrote result to {output_uri}")
187188
return [output_uri]
188189
except Exception as e:
189190
raise RuntimeError(f"Failed to write to remote storage: {e}")

‎sdk/python/feast/infra/compute_engines/ray/nodes.py‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -367,8 +367,7 @@ def execute(self, context: ExecutionContext) -> DAGValue:
367367
elif agg.function == "var":
368368
agg_dict[feature_name] = (agg.column, "var")
369369
else:
370-
logger.warning(f"Unknown aggregation function: {agg.function}")
371-
continue
370+
raise ValueError(f"Unknown aggregation function: {agg.function}.")
372371

373372
# Apply aggregations using pandas fallback (Ray's native groupby has compatibility issues)
374373
if self.group_by_keys and agg_dict:
@@ -422,8 +421,7 @@ def _fallback_pandas_aggregation(self, dataset: Dataset, agg_dict: dict) -> Data
422421
elif function == "var":
423422
result = grouped[column].var()
424423
else:
425-
logger.warning(f"Unknown aggregation function: {function}")
426-
continue
424+
raise ValueError(f"Unknown aggregation function: {function}.")
427425

428426
result.name = feature_name
429427
agg_results.append(result)
@@ -680,7 +678,7 @@ def write_batch_with_serialized_artifacts(batch: pd.DataFrame) -> pd.DataFrame:
680678
if getattr(feature_view, "online", False):
681679
# TODO: Implement proper online store writing with correct data format conversion
682680
logger.debug(
683-
f"Online store writing not implemented yet for {len(batch)} rows"
681+
"Online store writing not implemented yet for Ray compute engine"
684682
)
685683

686684
# Write to offline store if enabled

‎sdk/python/feast/infra/offline_stores/contrib/ray_offline_store/ray.py‎

Lines changed: 13 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -48,9 +48,13 @@
4848
from feast.on_demand_feature_view import OnDemandFeatureView
4949
from feast.repo_config import FeastConfigBaseModel, RepoConfig
5050
from feast.saved_dataset import SavedDatasetStorage, ValidationReference
51-
from feast.type_map import feast_value_type_to_pandas_type, pa_to_feast_value_type
51+
from feast.type_map import (
52+
convert_array_column,
53+
convert_scalar_column,
54+
feast_value_type_to_pandas_type,
55+
pa_to_feast_value_type,
56+
)
5257
from feast.utils import _get_column_names, make_df_tzaware, make_tzaware
53-
from feast.value_type import ValueType
5458

5559
logger = logging.getLogger(__name__)
5660

@@ -284,12 +288,12 @@ def convert_batch(batch: pd.DataFrame) -> pd.DataFrame:
284288
try:
285289
value_type = feature.dtype.to_value_type()
286290
if value_type.name.endswith("_LIST"):
287-
batch[feat_name] = _convert_array_column(
291+
batch[feat_name] = convert_array_column(
288292
batch[feat_name], value_type
289293
)
290294
else:
291295
target_pandas_type = feast_value_type_to_pandas_type(value_type)
292-
batch[feat_name] = _convert_scalar_column(
296+
batch[feat_name] = convert_scalar_column(
293297
batch[feat_name], value_type, target_pandas_type
294298
)
295299
except Exception as e:
@@ -302,64 +306,19 @@ def convert_batch(batch: pd.DataFrame) -> pd.DataFrame:
302306
return _apply_to_data(data, convert_batch)
303307

304308

305-
def _convert_scalar_column(
306-
series: pd.Series, value_type: ValueType, target_pandas_type: str
307-
) -> pd.Series:
308-
"""Convert a scalar feature column to the appropriate pandas type."""
309-
if value_type == ValueType.INT32:
310-
return pd.to_numeric(series, errors="coerce").astype("Int32")
311-
elif value_type == ValueType.INT64:
312-
return pd.to_numeric(series, errors="coerce").astype("Int64")
313-
elif value_type in [ValueType.FLOAT, ValueType.DOUBLE]:
314-
return pd.to_numeric(series, errors="coerce").astype("float64")
315-
elif value_type == ValueType.BOOL:
316-
return series.astype("boolean")
317-
elif value_type == ValueType.STRING:
318-
return series.astype("string")
319-
elif value_type == ValueType.UNIX_TIMESTAMP:
320-
return pd.to_datetime(series, unit="s", errors="coerce")
321-
else:
322-
return series.astype(target_pandas_type)
323-
324-
325-
def _convert_array_column(series: pd.Series, value_type: ValueType) -> pd.Series:
326-
"""Convert an array feature column to the appropriate type with proper empty array handling."""
327-
base_type_map = {
328-
ValueType.INT32_LIST: np.int32,
329-
ValueType.INT64_LIST: np.int64,
330-
ValueType.FLOAT_LIST: np.float32,
331-
ValueType.DOUBLE_LIST: np.float64,
332-
ValueType.BOOL_LIST: np.bool_,
333-
ValueType.STRING_LIST: object,
334-
ValueType.BYTES_LIST: object,
335-
ValueType.UNIX_TIMESTAMP_LIST: "datetime64[s]",
336-
}
337-
338-
target_dtype = base_type_map.get(value_type, object)
339-
340-
def convert_array_item(item) -> Union[np.ndarray, Any]:
341-
if item is None or (isinstance(item, list) and len(item) == 0):
342-
if target_dtype == object:
343-
return np.empty(0, dtype=object)
344-
else:
345-
return np.empty(0, dtype=target_dtype)
346-
else:
347-
return item
348-
349-
return series.apply(convert_array_item)
350-
351-
352309
class RayOfflineStoreConfig(FeastConfigBaseModel):
353310
"""
354311
Configuration for the Ray Offline Store.
312+
313+
For detailed configuration options and examples, see the documentation:
314+
https://docs.feast.dev/reference/offline-stores/ray
355315
"""
356316

357317
type: Literal[
358318
"feast.offline_stores.contrib.ray_offline_store.ray.RayOfflineStore", "ray"
359319
] = "ray"
360320
storage_path: Optional[str] = None
361321
ray_address: Optional[str] = None
362-
use_ray_cluster: Optional[bool] = False
363322

364323
# Optimization settings
365324
broadcast_join_threshold_mb: Optional[int] = 100
@@ -378,6 +337,7 @@ class RayOfflineStoreConfig(FeastConfigBaseModel):
378337
class RayResourceManager:
379338
"""
380339
Manages Ray cluster resources for optimal performance.
340+
# See: https://docs.feast.dev/reference/offline-stores/ray#resource-management-and-testing
381341
"""
382342

383343
def __init__(self, config: Optional[RayOfflineStoreConfig] = None) -> None:
@@ -1265,7 +1225,7 @@ def _ensure_ray_initialized(config: Optional[RepoConfig] = None) -> None:
12651225

12661226
if config and hasattr(config, "offline_store"):
12671227
if isinstance(ray_config, RayOfflineStoreConfig):
1268-
if ray_config.use_ray_cluster and ray_config.ray_address:
1228+
if ray_config.ray_address:
12691229
ray_init_kwargs["address"] = ray_config.ray_address
12701230
else:
12711231
ray_init_kwargs.update(

‎sdk/python/feast/infra/offline_stores/contrib/ray_repo_configuration.py‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,6 @@ def __init__(self, project_name: str, *args, **kwargs):
3030
type="ray",
3131
storage_path="/tmp/ray-storage",
3232
ray_address=None,
33-
use_ray_cluster=False,
3433
broadcast_join_threshold_mb=25,
3534
max_parallelism_multiplier=1,
3635
target_partition_size_mb=16,

0 commit comments

Comments
 (0)