Skip to content

Commit cf0e01f

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

12 files changed

Lines changed: 25 additions & 54 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: 2 additions & 2 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)

‎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/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: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -342,7 +342,7 @@ def convert_array_item(item) -> Union[np.ndarray, Any]:
342342
if target_dtype == object:
343343
return np.empty(0, dtype=object)
344344
else:
345-
return np.empty(0, dtype=target_dtype)
345+
return np.empty(0, dtype=target_dtype) # type: ignore
346346
else:
347347
return item
348348

@@ -359,7 +359,6 @@ class RayOfflineStoreConfig(FeastConfigBaseModel):
359359
] = "ray"
360360
storage_path: Optional[str] = None
361361
ray_address: Optional[str] = None
362-
use_ray_cluster: Optional[bool] = False
363362

364363
# Optimization settings
365364
broadcast_join_threshold_mb: Optional[int] = 100
@@ -1265,7 +1264,7 @@ def _ensure_ray_initialized(config: Optional[RepoConfig] = None) -> None:
12651264

12661265
if config and hasattr(config, "offline_store"):
12671266
if isinstance(ray_config, RayOfflineStoreConfig):
1268-
if ray_config.use_ray_cluster and ray_config.ray_address:
1267+
if ray_config.ray_address:
12691268
ray_init_kwargs["address"] = ray_config.ray_address
12701269
else:
12711270
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,

‎sdk/python/feast/infra/ray_shared_utils.py‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -141,7 +141,6 @@ def join_batch_with_features(batch: pd.DataFrame) -> pd.DataFrame:
141141
col for col in feature_cols if col in features.columns
142142
]
143143
features_filtered = features[available_feature_cols].copy()
144-
from .ray_shared_utils import normalize_timestamp_columns
145144

146145
batch = normalize_timestamp_columns(batch, timestamp_field, inplace=True)
147146
features_filtered = normalize_timestamp_columns(

‎sdk/python/tests/integration/compute_engines/ray_compute/repo_configuration.py‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@ def get_ray_compute_engine_test_config() -> IntegrationTestRepoConfig:
1919
offline_store_creator=RayDataSourceCreator,
2020
batch_engine={
2121
"type": "ray.engine",
22-
"use_ray_cluster": False,
2322
"max_workers": 1,
2423
"enable_optimization": True,
2524
},

0 commit comments

Comments
 (0)