diff --git a/docs/getting-started/components/compute-engine.md b/docs/getting-started/components/compute-engine.md index baf031b4284..60da1575932 100644 --- a/docs/getting-started/components/compute-engine.md +++ b/docs/getting-started/components/compute-engine.md @@ -25,7 +25,7 @@ engines. | SnowflakeComputeEngine | Runs on Snowflake, designed for scalable feature generation using Snowflake SQL. | ✅ | | | LambdaComputeEngine | Runs on AWS Lambda, designed for serverless feature generation. | ✅ | | | FlinkComputeEngine | Runs on Apache Flink, designed for stream processing and real-time feature generation. | ❌ | | -| RayComputeEngine | Runs on Ray, designed for distributed feature generation and machine learning workloads. | ❌ | | +| RayComputeEngine | Runs on Ray, designed for distributed feature generation and machine learning workloads. | ✅ | | ``` ### Batch Engine diff --git a/docs/getting-started/genai.md b/docs/getting-started/genai.md index 9c8f0c955d2..b4bdf1d1dc8 100644 --- a/docs/getting-started/genai.md +++ b/docs/getting-started/genai.md @@ -104,6 +104,24 @@ This integration enables: - Efficiently materializing features to vector databases - Scaling RAG applications to enterprise-level document repositories +### Scaling with Ray Integration + +Feast integrates with Ray to enable distributed processing for RAG applications: + +* **Ray Compute Engine**: Distributed feature computation using Ray's task and actor model +* **Ray Offline Store**: Process large document collections and generate embeddings at scale +* **Ray Batch Materialization**: Efficiently materialize features from offline to online stores +* **Distributed Embedding Generation**: Scale embedding generation across multiple nodes + +This integration enables: +- Distributed processing of large document collections +- Parallel embedding generation for millions of text chunks +- Kubernetes-native scaling for RAG applications +- Efficient resource utilization across multiple nodes +- Production-ready distributed RAG pipelines + +For detailed information on building distributed RAG applications with Feast and Ray, see [Feast + Ray: Distributed Processing for RAG Applications](https://feast.dev/blog/feast-ray-distributed-processing/). + ## Model Context Protocol (MCP) Support Feast supports the Model Context Protocol (MCP), which enables AI agents and applications to interact with your feature store through standardized MCP interfaces. This allows seamless integration with LLMs and AI agents for GenAI applications. @@ -158,6 +176,7 @@ For more detailed information and examples: * [RAG Tutorial with Docling](../tutorials/rag-with-docling.md) * [RAG Fine Tuning with Feast and Milvus](../../examples/rag-retriever/README.md) * [Milvus Quickstart Example](https://github.com/feast-dev/feast/tree/master/examples/rag/milvus-quickstart.ipynb) +* [Feast + Ray: Distributed Processing for RAG Applications](https://feast.dev/blog/feast-ray-distributed-processing/) * [MCP Feature Store Example](../../examples/mcp_feature_store/) * [MCP Feature Server Reference](../reference/feature-servers/mcp-feature-server.md) * [Spark Data Source](../reference/data-sources/spark.md) diff --git a/docs/reference/compute-engine/ray.md b/docs/reference/compute-engine/ray.md index 5547901b873..22b1e1a4700 100644 --- a/docs/reference/compute-engine/ray.md +++ b/docs/reference/compute-engine/ray.md @@ -2,6 +2,24 @@ 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: @@ -365,6 +383,8 @@ batch_engine: ### With Feature Transformations +#### On-Demand Transformations + ```python from feast import FeatureView, Field from feast.types import Float64 @@ -385,4 +405,27 @@ features = store.get_historical_features( ) ``` +#### 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). \ No newline at end of file diff --git a/docs/reference/offline-stores/README.md b/docs/reference/offline-stores/README.md index ab25fe9a276..b5e2bccbdd1 100644 --- a/docs/reference/offline-stores/README.md +++ b/docs/reference/offline-stores/README.md @@ -45,3 +45,7 @@ Please see [Offline Store](../../getting-started/components/offline-store.md) fo {% content-ref url="mssql.md" %} [mssql.md](mssql.md) {% endcontent-ref %} + +{% content-ref url="ray.md" %} +[ray.md](ray.md) +{% endcontent-ref %} diff --git a/docs/reference/offline-stores/overview.md b/docs/reference/offline-stores/overview.md index 191ccd21a64..24d37da22f1 100644 --- a/docs/reference/offline-stores/overview.md +++ b/docs/reference/offline-stores/overview.md @@ -26,33 +26,33 @@ The first three of these methods all return a `RetrievalJob` specific to an offl ## Functionality Matrix There are currently four core offline store implementations: `DaskOfflineStore`, `BigQueryOfflineStore`, `SnowflakeOfflineStore`, and `RedshiftOfflineStore`. -There are several additional implementations contributed by the Feast community (`PostgreSQLOfflineStore`, `SparkOfflineStore`, and `TrinoOfflineStore`), which are not guaranteed to be stable or to match the functionality of the core implementations. +There are several additional implementations contributed by the Feast community (`PostgreSQLOfflineStore`, `SparkOfflineStore`, `TrinoOfflineStore`, and `RayOfflineStore`), which are not guaranteed to be stable or to match the functionality of the core implementations. Details for each specific offline store, such as how to configure it in a `feature_store.yaml`, can be found [here](README.md). Below is a matrix indicating which offline stores support which methods. -| | Dask | BigQuery | Snowflake | Redshift | Postgres | Spark | Trino | Couchbase | -| :-------------------------------- | :-- | :-- | :-- | :-- | :-- | :-- | :-- | :-- | -| `get_historical_features` | yes | yes | yes | yes | yes | yes | yes | yes | -| `pull_latest_from_table_or_query` | yes | yes | yes | yes | yes | yes | yes | yes | -| `pull_all_from_table_or_query` | yes | yes | yes | yes | yes | yes | yes | yes | -| `offline_write_batch` | yes | yes | yes | yes | no | no | no | no | -| `write_logged_features` | yes | yes | yes | yes | no | no | no | no | +|| | Dask | BigQuery | Snowflake | Redshift | Postgres | Spark | Trino | Couchbase | Ray | +|| :-------------------------------- | :-- | :-- | :-- | :-- | :-- | :-- | :-- | :-- | :-- | +|| `get_historical_features` | yes | yes | yes | yes | yes | yes | yes | yes | yes | +|| `pull_latest_from_table_or_query` | yes | yes | yes | yes | yes | yes | yes | yes | yes | +|| `pull_all_from_table_or_query` | yes | yes | yes | yes | yes | yes | yes | yes | yes | +|| `offline_write_batch` | yes | yes | yes | yes | no | no | no | no | yes | +|| `write_logged_features` | yes | yes | yes | yes | no | no | no | no | yes | Below is a matrix indicating which `RetrievalJob`s support what functionality. -| | Dask | BigQuery | Snowflake | Redshift | Postgres | Spark | Trino | DuckDB | Couchbase | -| --------------------------------- | --- | --- | --- | --- | --- | --- | --- | --- | --- | -| export to dataframe | yes | yes | yes | yes | yes | yes | yes | yes | yes | -| export to arrow table | yes | yes | yes | yes | yes | yes | yes | yes | yes | -| export to arrow batches | no | no | no | yes | no | no | no | no | no | -| export to SQL | no | yes | yes | yes | yes | no | yes | no | yes | -| export to data lake (S3, GCS, etc.) | no | no | yes | no | yes | no | no | no | yes | -| export to data warehouse | no | yes | yes | yes | yes | no | no | no | yes | -| export as Spark dataframe | no | no | yes | no | no | yes | no | no | no | -| local execution of Python-based on-demand transforms | yes | yes | yes | yes | yes | no | yes | yes | yes | -| remote execution of Python-based on-demand transforms | no | no | no | no | no | no | no | no | no | -| persist results in the offline store | yes | yes | yes | yes | yes | yes | no | yes | yes | -| preview the query plan before execution | yes | yes | yes | yes | yes | yes | yes | no | yes | -| read partitioned data | yes | yes | yes | yes | yes | yes | yes | yes | yes | +|| | Dask | BigQuery | Snowflake | Redshift | Postgres | Spark | Trino | DuckDB | Couchbase | Ray | +|| --------------------------------- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | +|| export to dataframe | yes | yes | yes | yes | yes | yes | yes | yes | yes | yes | +|| export to arrow table | yes | yes | yes | yes | yes | yes | yes | yes | yes | yes | +|| export to arrow batches | no | no | no | yes | no | no | no | no | no | no | +|| export to SQL | no | yes | yes | yes | yes | no | yes | no | yes | no | +|| export to data lake (S3, GCS, etc.) | no | no | yes | no | yes | no | no | no | yes | yes | +|| export to data warehouse | no | yes | yes | yes | yes | no | no | no | yes | no | +|| export as Spark dataframe | no | no | yes | no | no | yes | no | no | no | no | +|| local execution of Python-based on-demand transforms | yes | yes | yes | yes | yes | no | yes | yes | yes | yes | +|| remote execution of Python-based on-demand transforms | no | no | no | no | no | no | no | no | no | no | +|| persist results in the offline store | yes | yes | yes | yes | yes | yes | no | yes | yes | yes | +|| preview the query plan before execution | yes | yes | yes | yes | yes | yes | yes | no | yes | yes | +|| read partitioned data | yes | yes | yes | yes | yes | yes | yes | yes | yes | yes | \ No newline at end of file diff --git a/docs/reference/offline-stores/ray.md b/docs/reference/offline-stores/ray.md index a46102ee132..89040f05c94 100644 --- a/docs/reference/offline-stores/ray.md +++ b/docs/reference/offline-stores/ray.md @@ -5,6 +5,23 @@ The Ray offline store is a data I/O implementation that leverages [Ray](https://www.ray.io/) for reading and writing data from various sources. It focuses on efficient data access operations, while complex feature computation is handled by the [Ray Compute Engine](../compute-engine/ray.md). +## Quick Start with Ray Template + +The easiest way to get started with Ray offline store is to use the built-in Ray template: + +```bash +feast init -t ray my_ray_project +cd my_ray_project/feature_repo +``` + +This template includes: +- Pre-configured Ray offline store and compute engine setup +- Sample feature definitions optimized for Ray processing +- Demo workflow showcasing Ray capabilities +- Resource settings for local development + +The template provides a complete working example with sample datasets and demonstrates both Ray offline store data I/O operations and Ray compute engine distributed processing. + ## Overview The Ray offline store provides: diff --git a/infra/website/docs/blog/feast-ray-distributed-processing.md b/infra/website/docs/blog/feast-ray-distributed-processing.md new file mode 100644 index 00000000000..cc637da4e68 --- /dev/null +++ b/infra/website/docs/blog/feast-ray-distributed-processing.md @@ -0,0 +1,226 @@ +--- +title: "Scaling ML with Feast and Ray: Distributed Processing for Modern AI Applications" +description: "Learn how Feast's integration with Ray enables distributed processing for both traditional feature engineering and modern RAG applications, with support for Kubernetes deployment through KubeRay." +date: 2025-10-29 +authors: ["Nikhil Kathole"] +--- + +
+ Feast + Ray Architecture for Distributed Processing +
+ +In today's data-driven world, organizations are increasingly turning to distributed computing to handle large-scale machine learning workloads. When it comes to feature engineering and retrieval-augmented generation (RAG) systems, the combination of **Feast** and **Ray** provides a powerful solution for building scalable, production-ready pipelines. + +This blog post explores how Feast's integration with Ray enables distributed processing for both traditional feature engineering and modern RAG applications, with support for Kubernetes deployment through KubeRay. + +## The Scaling Challenge + +Modern ML teams face critical scaling challenges: + +- **Massive Datasets**: Processing millions of documents for embedding generation +- **Complex Transformations**: CPU-intensive operations like text processing and feature engineering +- **Real-time Requirements**: Low-latency retrieval for RAG applications +- **Resource Efficiency**: Optimal utilization of compute resources across clusters + +## Building Scalable Feature Pipelines and RAG Systems with Distributed Computing + +Feast's integration with Ray addresses these challenges head-on, providing a unified platform where distributed processing is the default, not an afterthought. The magic happens when you realize that embedding generation, one of the most computationally intensive tasks in modern AI, can be treated as just another transformation in your feature pipeline. + +### The Ray RAG Revolution + +Consider the Ray RAG template, which demonstrates this new approach in action: + +```bash +# Built-in RAG template with distributed embedding generation +feast init -t ray_rag my_rag_project +cd my_rag_project/feature_repo +``` + +This single command gives you a complete system that can process thousands of documents in parallel, generate embeddings using distributed computing, and serve them through a vector database. + +The Ray RAG template demonstrates: + +- **Parallel Embedding Generation**: Distribute embedding computation across workers +- **Vector Search Integration**: Seamless integration with vector databases for similarity search +- **Complete RAG Pipeline**: Data → Embeddings → Search in one workflow + +## Embedding Generation as a Feast Transformation + +Feast's Ray integration makes embedding generation a first-class transformation operation. When you define a transformation in Feast, Ray handles the complexity of distributed processing. It partitions your data, distributes the computation across available workers, and manages the orchestration, all transparently to the developer. Here's how it works in practice: + +### Distributed Embedding Processing + +```python +from feast import BatchFeatureView, Entity, Field, FileSource +from feast.types import Array, Float32, String +from datetime import timedelta + +# Embedding processor for distributed Ray processing +class EmbeddingProcessor: + """Generate embeddings using SentenceTransformer model.""" + + def __init__(self): + import torch + from sentence_transformers import SentenceTransformer + + device = "cuda" if torch.cuda.is_available() else "cpu" + self.model = SentenceTransformer("all-MiniLM-L6-v2", device=device) + + def __call__(self, batch): + """Process batch and generate embeddings.""" + descriptions = batch["Description"].fillna("").tolist() + embeddings = self.model.encode( + descriptions, + show_progress_bar=False, + batch_size=128, + normalize_embeddings=True, + convert_to_numpy=True, + ) + batch["embedding"] = embeddings.tolist() + return batch + +# Ray native UDF for distributed processing +def generate_embeddings_ray_native(ds): + """Distributed embedding generation using Ray Data.""" + max_workers = 8 + batch_size = 2500 + + # Optimize partitioning for available workers + num_blocks = ds.num_blocks() + if num_blocks < max_workers: + ds = ds.repartition(max_workers) + + result = ds.map_batches( + EmbeddingProcessor, + batch_format="pandas", + concurrency=max_workers, + batch_size=batch_size, + ) + return result + +# Feature view with Ray transformation +document_embeddings_view = BatchFeatureView( + name="document_embeddings", + entities=[document], + mode="ray", # Native Ray Dataset mode + ttl=timedelta(days=365 * 100), + 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, +) +``` + +### RAG Query Example + +```python +from feast import FeatureStore +from sentence_transformers import SentenceTransformer + +# Initialize feature store +store = FeatureStore(repo_path=".") + +# Generate query embedding +model = SentenceTransformer("all-MiniLM-L6-v2") +query_embedding = model.encode(["sci-fi movie about space"])[0].tolist() + +# Retrieve similar documents +results = store.retrieve_online_documents_v2( + features=[ + "document_embeddings:embedding", + "document_embeddings:movie_name", + "document_embeddings:movie_director", + ], + query=query_embedding, + top_k=5, +).to_dict() + +# Display results +for i in range(len(results["document_id_pk"])): + print(f"{i+1}. {results['movie_name'][i]}") + print(f" Director: {results['movie_director'][i]}") + print(f" Distance: {results['distance'][i]:.3f}") +``` + +## Component Responsibilities + +The Feast + Ray integration follows a clear separation of concerns: + +- **Ray Compute Engine**: Executes distributed feature computations, transformations, and joins +- **Ray Offline Store**: Handles data I/O operations, reading from various sources (Parquet, CSV, etc.) + +This architectural separation ensures that each component has a single responsibility, making the system more maintainable and allowing for independent optimization of data access and computation layers. + +## Ray Integration Modes + +Feast supports three execution modes for Ray integration: + +### 1. Local Development +Perfect for experimentation and testing: + +```yaml +offline_store: + type: ray + storage_path: data/ray_storage + # Conservative settings for local development + broadcast_join_threshold_mb: 25 + max_parallelism_multiplier: 1 + target_partition_size_mb: 16 +``` + +### 2. Remote Ray Cluster +Connect to existing Ray infrastructure: + +```yaml +offline_store: + type: ray + storage_path: s3://my-bucket/feast-data + ray_address: "ray://my-cluster.example.com:10001" +``` + +### 3. Kubernetes with KubeRay +Enterprise-ready deployment: + +```yaml +offline_store: + type: ray + storage_path: s3://my-bucket/feast-data + use_kuberay: true + kuberay_conf: + cluster_name: "feast-ray-cluster" + namespace: "feast-system" +``` + +## Getting Started + +### Install Feast with Ray Support +```bash +pip install feast[ray] +``` + +### Initialize Ray RAG Template +```bash +# RAG applications with distributed embedding generation +feast init -t ray_rag my_rag_project +cd my_rag_project/feature_repo +``` + +### Deploy to Production +```bash +feast apply +feast materialize --disable-event-timestamp +python test_workflow.py +``` + +Whether you're building traditional feature pipelines or modern RAG systems, Feast + Ray offers the scalability and performance needed for production workloads. The integration supports everything from local development to large-scale Kubernetes deployments, making it an ideal choice for organizations looking to scale their ML infrastructure. + +--- + +**Ready to build distributed RAG applications?** Get started with our [Ray RAG template](https://docs.feast.dev/reference/compute-engine/ray) and explore [Feast + Ray documentation](https://docs.feast.dev/reference/offline-stores/ray) for distributed embedding generation. + +*Learn more about Feast's distributed processing capabilities and join the community at [feast.dev](https://feast.dev).* diff --git a/infra/website/public/images/blog/feast_ray_architecture.png b/infra/website/public/images/blog/feast_ray_architecture.png new file mode 100644 index 00000000000..649d912a790 Binary files /dev/null and b/infra/website/public/images/blog/feast_ray_architecture.png differ