# Ray (contrib) The Ray compute engine is a distributed compute implementation that leverages [Ray](https://www.ray.io/) for executing feature pipelines including transformations, aggregations, joins, and materializations. It provides scalable and efficient distributed processing for both `materialize()` and `get_historical_features()` operations. ## Quick Start with Ray Template ### Ray RAG Template - Batch Embedding at Scale For RAG (Retrieval-Augmented Generation) applications with distributed embedding generation: ```bash feast init -t ray_rag my_rag_project cd my_rag_project/feature_repo ``` The Ray RAG template demonstrates: * **Parallel Embedding Generation**: Uses Ray compute engine to generate embeddings across multiple workers * **Vector Search Integration**: Works with Milvus for semantic similarity search * **Complete RAG Pipeline**: Data → Embeddings → Search workflow The Ray compute engine automatically distributes the embedding generation across available workers, making it ideal for processing large datasets efficiently. ## Overview The Ray compute engine provides: * **Distributed DAG Execution**: Executes feature computation DAGs across Ray clusters * **Intelligent Join Strategies**: Automatic selection between broadcast and distributed joins * **Lazy Evaluation**: Deferred execution for optimal performance * **Resource Management**: Automatic scaling and resource optimization * **Point-in-Time Joins**: Efficient temporal joins for historical feature retrieval ## Architecture The Ray compute engine follows Feast's DAG-based architecture: ``` EntityDF → RayReadNode → RayJoinNode → RayFilterNode → RayAggregationNode → RayTransformationNode → Output ``` ### Core Components | Component | Description | | ----------------------- | -------------------------------------------------- | | `RayComputeEngine` | Main engine implementing `ComputeEngine` interface | | `RayFeatureBuilder` | Constructs DAG from Feature View definitions | | `RayDAGNode` | Ray-specific DAG node implementations | | `RayDAGRetrievalJob` | Executes retrieval plans and returns results | | `RayMaterializationJob` | Handles materialization job tracking | ## Configuration Configure the Ray compute engine in your `feature_store.yaml`: ```yaml project: my_project registry: data/registry.db provider: local offline_store: type: ray storage_path: data/ray_storage batch_engine: type: ray.engine max_workers: 4 # Optional: Maximum number of workers enable_optimization: true # Optional: Enable performance optimizations broadcast_join_threshold_mb: 100 # Optional: Broadcast join threshold (MB) max_parallelism_multiplier: 2 # Optional: Parallelism multiplier target_partition_size_mb: 64 # Optional: Target partition size (MB) window_size_for_joins: "1H" # Optional: Time window for distributed joins ray_address: localhost:10001 # Optional: Ray cluster address ``` ### Configuration Options | Option | Type | Default | Description | | ----------------------------- | ------- | --------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `type` | string | `"ray.engine"` | Must be `ray.engine` | | `max_workers` | int | None (uses all cores) | Maximum number of Ray workers | | `enable_optimization` | boolean | true | Enable performance optimizations | | `broadcast_join_threshold_mb` | int | 100 | Size threshold for broadcast joins (MB) | | `max_parallelism_multiplier` | int | 2 | Parallelism as multiple of CPU cores | | `target_partition_size_mb` | int | 64 | Target partition size (MB) | | `window_size_for_joins` | string | "1H" | Time window for distributed joins | | `ray_address` | string | None | Ray cluster address (triggers REMOTE mode) | | `use_kuberay` | boolean | None | Enable KubeRay mode (overrides ray\_address) | | `kuberay_conf` | dict | None | **KubeRay configuration dict** with keys: `cluster_name` (required), `namespace` (default: "default"), `auth_token`, `auth_server`, `skip_tls` (default: false) | | `enable_ray_logging` | boolean | false | Enable Ray progress bars and logging | | `enable_distributed_joins` | boolean | true | Enable distributed joins for large datasets | | `staging_location` | string | None | Remote path for batch materialization jobs | | `ray_conf` | dict | None | Ray configuration parameters (memory, CPU limits) | ### Mode Detection Precedence The Ray compute engine automatically detects the execution mode: 1. **Environment Variables** → KubeRay mode (if `FEAST_RAY_USE_KUBERAY=true`) 2. **Config `kuberay_conf`** → KubeRay mode 3. **Config `ray_address`** → Remote mode 4. **Default** → Local mode ## Usage Examples ### Basic Historical Feature Retrieval ```python from feast import FeatureStore import pandas as pd from datetime import datetime # Initialize feature store with Ray compute engine store = FeatureStore("feature_store.yaml") # Create entity DataFrame entity_df = pd.DataFrame({ "driver_id": [1, 2, 3, 4, 5], "event_timestamp": [datetime.now()] * 5 }) # Get historical features using Ray compute engine features = store.get_historical_features( entity_df=entity_df, features=[ "driver_stats:avg_daily_trips", "driver_stats:total_distance" ] ) # Convert to DataFrame df = features.to_df() print(f"Retrieved {len(df)} rows with {len(df.columns)} columns") ``` ### Batch Materialization ```python from datetime import datetime, timedelta # Materialize features using Ray compute engine store.materialize( start_date=datetime.now() - timedelta(days=7), end_date=datetime.now(), feature_views=["driver_stats", "customer_stats"] ) # The Ray compute engine handles: # - Distributed data processing # - Optimal join strategies # - Resource management # - Progress tracking ``` ### Large-Scale Feature Retrieval ```python # Handle large entity datasets efficiently large_entity_df = pd.DataFrame({ "driver_id": range(1, 1000000), # 1M entities "event_timestamp": [datetime.now()] * 1000000 }) # Ray compute engine automatically: # - Partitions data optimally # - Selects appropriate join strategies # - Distributes computation across cluster features = store.get_historical_features( entity_df=large_entity_df, features=[ "driver_stats:avg_daily_trips", "driver_stats:total_distance", "customer_stats:lifetime_value" ] ).to_df() ``` ### Advanced Configuration ```yaml # Production-ready configuration batch_engine: type: ray.engine # Resource configuration max_workers: 16 max_parallelism_multiplier: 4 # Performance optimization enable_optimization: true broadcast_join_threshold_mb: 50 target_partition_size_mb: 128 # Distributed join configuration window_size_for_joins: "30min" # Ray cluster configuration ray_address: "ray://head-node:10001" ``` ### Complete Example Configuration Here's a complete example configuration showing how to use Ray offline store with Ray compute engine: ```yaml # Complete example configuration for Ray offline store + Ray compute engine # This shows how to use both components together for distributed processing project: my_feast_project registry: data/registry.db provider: local # Ray offline store configuration # Handles data I/O operations (reading/writing data) offline_store: type: ray storage_path: s3://my-bucket/feast-data # Optional: Path for storing datasets ray_address: localhost:10001 # Optional: Ray cluster address # Ray compute engine configuration # Handles complex feature computation and distributed processing batch_engine: type: ray.engine # Resource configuration max_workers: 8 # Maximum number of Ray workers max_parallelism_multiplier: 2 # Parallelism as multiple of CPU cores # Performance optimization enable_optimization: true # Enable performance optimizations broadcast_join_threshold_mb: 100 # Broadcast join threshold (MB) target_partition_size_mb: 64 # Target partition size (MB) # Distributed join configuration window_size_for_joins: "1H" # Time window for distributed joins # Ray cluster configuration (inherits from offline_store if not specified) ray_address: localhost:10001 # Ray cluster address ``` ## DAG Node Types The Ray compute engine implements several specialized DAG nodes: ### RayReadNode Reads data from Ray-compatible sources: * Supports Parquet, CSV, and other formats * Handles partitioning and schema inference * Applies field mappings and filters ### RayJoinNode Performs distributed joins: * **Broadcast Join**: For small datasets (<100MB) * **Distributed Join**: For large datasets with time-based windowing * **Automatic Strategy Selection**: Based on dataset size and cluster resources ### RayFilterNode Applies filters and time-based constraints: * TTL-based filtering * Timestamp range filtering * Custom predicate filtering ### RayAggregationNode Handles feature aggregations: * Windowed aggregations * Grouped aggregations * Custom aggregation functions ### RayTransformationNode Applies feature transformations: * Row-level transformations * Column-level transformations * Custom transformation functions ### RayWriteNode Writes results to various targets: * Online stores * Offline stores * Temporary storage ## Join Strategies The Ray compute engine automatically selects optimal join strategies: ### Broadcast Join Used for small feature datasets: * Automatically selected when feature data < 100MB * Features are cached in Ray's object store * Entities are distributed across cluster * Each worker gets a copy of feature data ### Distributed Windowed Join Used for large feature datasets: * Automatically selected when feature data > 100MB * Data is partitioned by time windows * Point-in-time joins within each window * Results are combined across windows ### Strategy Selection Logic ```python def select_join_strategy(feature_size_mb, threshold_mb): if feature_size_mb < threshold_mb: return "broadcast" else: return "distributed_windowed" ``` ## Performance Optimization ### Automatic Optimization The Ray compute engine includes several automatic optimizations: 1. **Partition Optimization**: Automatically determines optimal partition sizes 2. **Join Strategy Selection**: Chooses between broadcast and distributed joins 3. **Resource Allocation**: Scales workers based on available resources 4. **Memory Management**: Handles out-of-core processing for large datasets ### Manual Tuning For specific workloads, you can fine-tune performance: ```yaml batch_engine: type: ray.engine # Fine-tuning for high-throughput scenarios broadcast_join_threshold_mb: 200 # Larger broadcast threshold max_parallelism_multiplier: 1 # Conservative parallelism target_partition_size_mb: 512 # Larger partitions window_size_for_joins: "2H" # Larger time windows ``` ### Monitoring and Metrics Monitor Ray compute engine performance: ```python import ray # Check cluster resources resources = ray.cluster_resources() print(f"Available CPUs: {resources.get('CPU', 0)}") print(f"Available memory: {resources.get('memory', 0) / 1e9:.2f} GB") # Monitor job progress job = store.get_historical_features(...) # Ray compute engine provides built-in progress tracking ``` ## Integration Examples ### With Spark Offline Store ```yaml # Use Ray compute engine with Spark offline store offline_store: type: spark spark_conf: spark.executor.memory: "4g" spark.executor.cores: "2" batch_engine: type: ray.engine max_workers: 8 enable_optimization: true ``` ### With Cloud Storage ```yaml # Use Ray compute engine with cloud storage offline_store: type: ray storage_path: s3://my-bucket/feast-data batch_engine: type: ray.engine ray_address: "ray://ray-cluster:10001" broadcast_join_threshold_mb: 50 ``` ### With Feature Transformations #### On-Demand Transformations ```python from feast import FeatureView, Field from feast.types import Float64 from feast.on_demand_feature_view import on_demand_feature_view @on_demand_feature_view( sources=["driver_stats"], schema=[Field(name="trips_per_hour", dtype=Float64)] ) def trips_per_hour(features_df): features_df["trips_per_hour"] = features_df["avg_daily_trips"] / 24 return features_df # Ray compute engine handles transformations efficiently features = store.get_historical_features( entity_df=entity_df, features=["trips_per_hour:trips_per_hour"] ) ``` #### Ray Native Transformations For distributed transformations that leverage Ray's dataset and parallel processing capabilities, use `mode="ray"` in your `BatchFeatureView`: ```python # Feature view with Ray transformation mode document_embeddings_view = BatchFeatureView( name="document_embeddings", entities=[document], mode="ray", # Enable Ray native transformation ttl=timedelta(days=365), schema=[ Field(name="document_id", dtype=String), Field(name="embedding", dtype=Array(Float32), vector_index=True), Field(name="movie_name", dtype=String), Field(name="movie_director", dtype=String), ], source=movies_source, udf=generate_embeddings_ray_native, online=True, ) ``` For more information, see the [Ray documentation](https://docs.ray.io/en/latest/) and [Ray Data guide](https://docs.ray.io/en/latest/data/getting-started.html).