Billion-Scale Vector Search in Seconds with EMR Serverless Ray + Lance

Architecture Background: Connecting Multimodal Storage with Distributed Python Compute

In autonomous driving, image retrieval, and multimodal understanding, finding similar data is only the first step. Raw images require decoding, cleaning, and annotation; models generate embeddings; recall results need reranking and quality evaluation before entering training sets or production systems. These stages rely heavily on the Python data-processing and model ecosystem. When data grows to billions of records, the platform must support both large-scale storage with efficient retrieval and seamless execution of algorithm code on distributed CPU/GPU resources.

EMR Serverless Ray + Lance connects data storage, vector retrieval, and Python compute into a single workflow. Lance holds data and indexes; Lance-Ray bridges storage and compute; Ray handles distributed execution; EMR Serverless Ray provides a managed cluster and job runtime.

A leading Chinese autonomous-driving solution vendor tested this stack on a real billion-scale dataset and reported:

Pure vector retrieval P50: 1.23 seconds

Vector retrieval + multi-condition filtering P50: 3.95 seconds

Test configuration: 768-dimensional vectors, L2 distance, top-20,000, no cache, direct read from OSS. Figures represent single-query engine retrieval latency, not full result-set delivery.

1.1 Ray & EMR Serverless Ray: From Distributed Framework to Managed Platform

Ray is an open-source distributed compute framework for AI and Python apps. Core primitives: Task (parallel functions), Actor (stateful computation, e.g., model reuse across batches), and distributed object storage. Ray Data builds on these to provide data reading, transformation, and batch inference interfaces.

For image embedding generation, developers can use Python functions for decoding/preprocessing and an Actor holding the model for batch inference. With map_batches, the model loads once during Actor initialization and is reused across batches, avoiding repeated initialization overhead.

Ray Data uses streaming batch execution: data flows in chunks through processing stages. Backpressure limits upstream task submission when downstream lags, controlling intermediate data buildup. This underpins chaining reads, CPU preprocessing, and GPU inference; actual memory usage depends on batch size, parallelism, and model requirements.

EMR Serverless Ray adds production-grade managed compute on top of open-source Ray. It unifies Ray Cluster, Ray Job, Ray Serve, and History Server in one runtime, supporting multimodal data processing (images, video, audio, text) and model inference. Developers access resources via Notebook, job submission, SDK/API, or DataWorks scheduling.

Key platform capabilities for Lance multimodal workflows:

CPU/GPU hybrid scheduling & elasticity: Cross-queue scheduling within a single cluster, adjusting concurrency and scaling based on utilization and backlog; independent CPU/GPU scaling. Allocates data I/O, preprocessing, and model inference to appropriate resources, matching stage throughput.

Job observability & environment management: Centralized History Server retains historical dashboards and logs; supports cluster snapshot comparison, custom images, Prometheus, NVIDIA Nsight. Enables tracing long-running data-processing jobs, pinpointing bottlenecks in data reads, model compute, and resource usage.

Security, fault tolerance & recovery: Kerberos authentication, Ranger authorization; Ray Head fault tolerance, disk watermark detection, backpressure, GPU failure handling. Provides production guarantees for continuous batch processing and model serving.

Multimodal operators & model integration: Daft multimodal operators and Ray Data AI; supports folding model calls into data pipelines. Combines image codec, video frame extraction, content understanding, vector generation, and writes results back to Lance.

Storage integration & data acceleration: Connects OSS, CPFS, DLF, and Lance; plus independent cache clusters for faster data access. Accelerates repeated access in multi-round processing, training, and retrieval.

1.2 Lance: Columnar Storage Balancing Batch Scans and Random Access

Lance is an open-source columnar format for AI data. It organizes business attributes, vectors, and multimodal data in one dataset, balancing full scans with random access — suitable for batch embedding generation and post-retrieval candidate fetching. Via the Python package pylance, developers perform data I/O, retrieval, index building, and version management, linking storage to downstream model compute.

1.3 Lance-Ray: Bridging Data Access, Distributed Processing, and Result Write-Back

Lance-Ray plugs Lance datasets into Ray: read_lance yields a Ray Dataset; after processing, write_lance writes back. The integration adds filter pushdown, distributed column backfill, and data-file compaction, letting data preparation, model compute, and data maintenance run in one Python job.

Example: a job selects columns/records, generates new embeddings in Ray, and backfills results into the Lance dataset.

1.4 Overall Architecture

The architecture spans four layers: business entry, managed platform, heterogeneous compute, and data storage. CPU and GPU queues are organized per task; Lance-Ray connects Lance data with processing logic in Python jobs.

Lance-Ray Capabilities (v0.5.0 / pylance 10.0.0)

The table below summarizes community Lance-Ray capabilities; Section 3 only benchmarks vector retrieval performance.

Distributed read: read_lance with column selection, filtering, scan parameters — for reading business fields or model inputs from large Lance datasets.

Distributed write: write_lance with create, append, overwrite — for writing cleaned data, model outputs, vectors into Lance.

Python batch processing: Read as Ray Dataset, then use map_batches etc. — for image preprocessing, text cleaning, feature computation, batch inference.

New columns & distributed backfill: add_columns, add_columns_from, merge_columns_from — to supplement existing data with embeddings, labels, quality scores.

Vector index creation: create_index built by multiple workers — for large-scale vector indexing.

Scalar & full-text index creation: create_scalar_index — accelerates attribute filtering, builds full-text indexes.

Distributed vector search: vector_search merges global top-k — for image similarity search, semantic recall.

Incremental index maintenance: optimize_indices — maintains index coverage after appends.

Distributed file compaction: compact_files, compact_database — merges small fragments, improves scan efficiency.

Blob data I/O: Blob column read/write config — brings multimodal content into Python pipelines.

Namespace integration: Locate datasets via Namespace and table identifiers — connects to unified data catalog.

Vector index types exposed: IVF_FLAT, IVF_PQ, IVF_RQ, IVF_SQ, IVF_HNSW_FLAT, IVF_HNSW_PQ, IVF_HNSW_SQ. Scalar index string types: BTREE, BITMAP, INVERTED/FTS, ZONEMAP.

Important capability boundaries: optimize_indices currently delegates to pylance; not fully distributed. merge_columns_from is for column backfill, not a full upsert by business key.

FTS index build supported, but no global FTS ranking or hybrid search distributed interface yet.

Ray schedules GPU model tasks; GPU inference and Lance index build GPU support are separate concerns.

Billion-Scale Vector Retrieval Benchmark (Real Autonomous-Driving Data)

3.1 Data, Scenarios & Environment

Test data: CN_CLIP vectors from autonomous-driving collected images, 768 dimensions , L2 distance , target top-20,000 . Used 10e table.

Six query scenarios:

S1: Pure vector (no filter)

S2: Device ID filter

S3: Time-range filter

S4: Vehicle model filter

S5: Geo-range filter (longitude/latitude)

S6: Multi-condition (vehicle model, time, sensor, longitude, etc.)

Environment:

Core versions: Lance-Ray 0.5.0, pylance 10.0.0; batch parallel test used Ray 2.55.1

Cluster: 1 Head + 2–8 elastic Workers

Node spec: Head & Workers each 16 vCPU / 64 GB

Data access: Python jobs read OSS directly via pylance

Cache: No cache used , direct OSS reads

Query params: k=20,000, nprobes=40 Ray jobs organize Actors by vector index segment. The 10e table uses a 64-segment IVF_PQ index, executed by 16 Actors (each handling 4 segments). Actor-to-Worker mapping is decided by runtime scheduler; merged top-k is approximate, not guaranteed exact nearest neighbor or full recall.

3.2 Single-Query Latency (P50, seconds)

Engine retrieval scope only; excludes full business result-set delivery, so not end-to-end application response time.

S1 Pure vector: 1.23

S2 Device filter*: 1.40

S3 Time filter: 1.82

S4 Vehicle model filter: 1.33

S5 Geo filter: 2.71

S6 Multi-condition: 3.95

*S2 filter hit count below top-20,000; not valid as full-quota SLO evidence.

3.3 Batch Parallel Query Performance

Concurrency model: N queries form a batch, executed in parallel inside engine. Metric: batch elapsed time ÷ N = amortized per-query time. Amortized batch time ≠ single-request response time.

Three batch sizes tested: 20, 40, 80 queries. Ray batch processing shared-scan benefit range 1.4–1.8×, but no dedicated measurement for 10e table.

20-query batch amortized latency (seconds):

S1 Pure vector: 0.79

S2 Device filter*: 0.77

S3 Time filter: 0.83

S6 Multi-condition: 0.83

40/80 batch per-scenario latencies and 20-batch S4/S5 data incomplete; only available details shown.

Throughput: S1 pure vector, 80-query batch → 1.3 queries/second. Throughput stabilizes as batch grows from 20 to 80; batch elapsed time increases with batch size. Throughput variation across filter scenarios small; S6 close to S1.

Scenario differences & stability:

20-batch S1→S3: amortized time 0.79→0.83 sec

20-batch S6/S1 ratio: ~1.05× (0.83/0.79)

80-batch stability: full test matrix passed

3.4 Benchmark Summary

Without cache, reading OSS directly, EMR Serverless Ray + Lance achieves second-level engine retrieval on the 10e table. Single query covers pure vector to multi-filter; 20-batch amortized per-query time 0.77–0.83 sec with low variance across filters. As batch size grows, throughput stabilizes; future optimization can target resource utilization and data-access bottlenecks.

Adding EMR Serverless cache clusters can reuse hot data and indexes, reducing repeated OSS reads. Internal tests show 20%+ retrieval speedup with cache, benefiting hot queries and multi-round sample mining; 10e table tests did not enable cache, actual gain depends on hit rate and load.

On top of retrieval, candidate samples can flow into Python data-processing and model-compute pipelines for reranking, annotation, and result backfill, forming a complete multimodal processing chain.

Extending from Vector Retrieval to Python Multimodal Workflows

In image processing, model inference, and custom reranking scenarios, Ray + pylance unifies retrieval and algorithm code in one Python job, enabling unified development, distributed execution, and continuous iteration.

4.1 Reusing Python Data-Processing & Model Ecosystem

Algorithm teams' existing Python functions, models, and tools combine with pylance data access and retrieval interfaces, then scale across nodes via Ray Task, Actor, or Ray Data.

Query image preprocessing: Pillow, OpenCV — decode, crop, normalize before retrieval.

Generate query vectors / batch embeddings: PyTorch, Transformers, Sentence Transformers — load model, run inference, feed vectors to pylance retrieval or write back.

Candidate data processing: NumPy, Pandas, PyArrow — deduplicate, compute features, custom filtering on retrieval results.

Model reranking & business scoring: Custom PyTorch models, Cross-Encoder — rescore candidates, fuse similarity with business rules.

Multimodal sample mining: Vision models + custom evaluation code — read candidate images, generate labels, select training samples, write back to Lance.

4.2 Chaining Recall, Model Evaluation, and Sample Write-Back

Real business often needs to judge "which results are more valuable" after vector recall. Example: autonomous-driving team seeks "night, rain, construction obstacles" frames. Flow:

Vector recall narrows candidates.

Vision model identifies weather, lighting, obstacles.

Business rules drive reranking, deduplication, label write-back.

Developers can embed conditional branches, multi-round recall, and evaluation logic in Python, reducing adapters between query systems and algorithm jobs.

4.3 Organizing CPU, GPU, and Model Calls on Managed Platform

EMR Serverless Ray assigns Lance data reads and preprocessing to CPU queues, embedding generation and vision recognition to GPU queues. Cross-queue scheduling and independent scaling connect stage loads; Lance-Ray backfills vectors, labels, quality scores.

For content understanding, classification, tagging model API tasks, Ray Data AI uses ai_query on Dataset for batch calls: supports multimodal inputs (images, video frames), structured output, adaptive concurrency, OSS access handling, per-row error isolation and retry, and metrics like token consumption and execution latency.

Ray Data AI can also connect to Ray Serve LLM self-hosted model services: CPU-side ai_query sends OpenAI-compatible requests; GPU-side Ray Serve LLM orchestrates vLLM inference with flow control, service elasticity, load balancing. Existing EMR examples write results to Parquet; end-to-end combination with Lance-Ray read/write and column backfill still needs adaptation validation.

After model updates, teams reuse the processing pipeline to recompute results, tracking each run via History Server, logs, and resource monitoring.

4.4 Ray Serve + Lance: Reusing Model Services, Continuously Enriching Data

When scoring, tagging, or embedding models are repeatedly called by multiple tasks, deploy them as long-running services via Ray Serve. Data-processing jobs invoke them through DeploymentHandle in Python, composing multiple deployments into processing chains so model services and batch pipelines evolve independently.

Ray Serve + Lance enables image scoring and label backfill. Jobs read images and business fields from Lance in batches, call Ray Serve models for scores, labels, or vectors, then backfill results as new columns via Lance-Ray for downstream sample selection and multimodal training.

In autonomous driving, this pattern applies to image quality scoring, weather recognition, embedding generation. Different processing tasks reuse the same model service; model updates trigger recomputation and feature backfill on target data. Implementation must match model I/O, batching style, and Lance-Ray interface versions; this pattern was not part of Section 3 retrieval benchmarks.

4.5 From Batch Processing to Online Retrieval Service (Architecture Extension)

Using Ray Serve composition, one can design an online retrieval chain for Lance: receive request → call embedding deployment for query vector → retrieval deployment queries Lance via pylance → reranking deployment → return results. This is a to-be-validated architecture extension ; not yet end-to-end performance validated in this article's version or EMR environment.

Ray Serve handles request and service deployment orchestration; Lance provides data and index access. Adding Serve replicas ≠ automatic index sharding or cross-shard candidate merging — retrieval implementation must handle those. Deployment must also verify data version refresh, per-replica memory/cache overhead, and full-request-chain latency, throughput, recall; Section 3 engine retrieval numbers cannot directly serve as online service metrics.

Conclusion

This benchmark demonstrates EMR Serverless Ray + Lance achieves second-level retrieval on real autonomous-driving data , laying groundwork for subsequent sample mining and data processing. Vector retrieval finds candidates at scale; Python processing and model compute then assess sample value, turning retrieval results into labels, features, and training data.

The core value of this architecture is unifying retrieval and multimodal processing in a single Python workflow. Lance stores data and indexes; Lance-Ray bridges distributed read/write and result backfill; EMR Serverless Ray supplies managed runtime, elastic resources, and job observability. Teams can reuse existing models and algorithm code, progressively linking data cleaning, vector generation, candidate recall, reranking/scoring, and sample construction. For repeatedly used models, Ray Serve offers shared services, supporting continuous data processing and model iteration.

EMR Serverless cache clusters add further acceleration headroom. Reusing hot data and indexes reduces OSS read overhead, shortens data load time, improves retrieval latency, and continuously boosts retrieval and multimodal processing efficiency for multi-round training, repeated debugging, and model evaluation.

Starting from validated vector retrieval, incrementally layering model compute, service reuse, and data acceleration, EMR Serverless Ray + Lance provides a sustainably evolvable implementation path for multimodal applications, letting the same data serve retrieval, training, and business decisions.