Architecture of Yelp’s In-House Search Engine: nrtSearch

Elasticsearch is one of the most widely adopted distributed search engines in modern software architectures. However, at Yelp’s operational scale—where millions of users discover local businesses and read reviews in real time—Elasticsearch began exhibiting architectural bottlenecks around replication, resource utilization, and operational complexity.

To overcome these limitations, Yelp engineered nrtSearch (Near Real-Time Search): a custom search engine built directly on top of Apache Lucene. By rethinking replication, indexing, and communication protocols, Yelp achieved lower latency, higher throughput, and simpler operational maintenance.

---

Why Yelp Replaced Elasticsearch

At its core, Elasticsearch is a distributed abstraction layer over Apache Lucene. Lucene is an in-process Java library handling low-level inverted index construction, segment merging, scoring, and query execution. Elasticsearch turns Lucene into a distributed, multi-tenant, HTTP-accessible search engine.

While this abstraction provides rich developer convenience, it introduces several critical trade-offs at high throughput:

1. Document-Based Replication Overhead

In Elasticsearch, write replication happens at the document operation level:

• When a document is written, the write query hits the primary shard.
• The primary shard coordinates the write and forwards the raw document payload or index operation to every replica shard.
• Each replica shard independently performs text tokenization, analysis, stem evaluation, and inverted index computation.

                    [ Write Request ]
                            │
                            ▼
                 ┌─────────────────────┐
                 │    Primary Shard    │
                 │ (Parses & Indexes)  │
                 └──────────┬──────────┘
                            │
             ┌──────────────┴──────────────┐
             ▼                             ▼
  ┌─────────────────────┐       ┌─────────────────────┐
  │    Replica Shard    │       │    Replica Shard    │
  │ (Parses & Indexes)  │       │ (Parses & Indexes)  │
  └─────────────────────┘       └─────────────────────┘

Because every replica must duplicate the heavy CPU-bound tasks of tokenization and indexing, scaling read replicas requires paying an equally high CPU tax for indexing. You cannot simply scale out with lightweight, read-optimized compute instances.

2. Uneven Load Distribution and Hot Spots

Elasticsearch decouples logical shards from physical data nodes. An index might have 5 logical shards distributed unevenly across 3 nodes. If queries or indexing traffic hit specific shards disproportionately, those physical nodes become hot spots.

When a physical data node experiences resource starvation, operators often need to resort to manual rebalancing or third-party tooling (such as Elasticsearch Head) to relocate shards across physical nodes.

3. Challenging Auto-Scaling and Over-Provisioning

Auto-scaling an Elasticsearch cluster dynamically in response to traffic spikes is notoriously difficult:

• Spinning up a new Elasticsearch node requires cluster rebalancing, network-heavy shard allocation, and synchronization.
• Rebalancing actively competes with production query traffic for I/O and network bandwidth.
• Consequently, teams are forced to permanently over-provision for peak traffic, leaving expensive CPU and RAM idle during off-peak hours.

---

The Foundation of nrtSearch

Yelp realized they did not need many of Elasticsearch’s enterprise features—such as integrated time-series aggregations, complex analytics, or built-in kibana abstractions. What they needed was near-real-time index freshness, low query latencies, and predictable horizontal scaling.

They based their work on the open-source Lucene Server project (originally created by Michael McCandless, a core Lucene developer) and built an enterprise-grade search system: nrtSearch.

┌───────────────────────────────────────────────┐
│         Client / Application Layer            │
└───────────────┬───────────────────────────────┘
                │ gRPC (Protobuf) / REST (via gRPC Gateway)
┌───────────────▼───────────────────────────────┐
│                 nrtSearch                     │
│  ┌─────────────────────────────────────────┐  │
│  │ Virtual Sharding & Slicing Search Pool  │  │
│  └─────────────────────────────────────────┘  │
│  ┌─────────────────────────────────────────┐  │
│  │ Segment-Based Replication Engine        │  │
│  └─────────────────────────────────────────┘  │
│  ┌─────────────────────────────────────────┐  │
│  │ Apache Lucene Engine (Inverted Indexes) │  │
│  └─────────────────────────────────────────┘  │
└───────────────────────────────────────────────┘

Core Lucene Capabilities Leveraged

1. Immutable Segments & Segment-Based Replication

In Apache Lucene, data is organized into immutable on-disk files known as segments. When new documents are flushed to disk, a new immutable segment is produced.

Because segments are write-once and never modified, a replica does not need to parse or tokenize raw documents. Instead, the replica can copy the pre-built, finalized segment directly from the primary node. Once the segment file is copied locally, the replica points its searcher to the new segment and immediately begins serving queries.

2. Segment-Level Concurrent Search

Lucene evaluates queries per segment before aggregating final relevance scores. Rather than executing a search query sequentially across an entire node, Lucene allows parallel execution across multiple distinct segments concurrently, fully utilizing multicore CPU architectures.

---

Key Architectural Decisions in nrtSearch

1. High-Performance gRPC Communication

Elasticsearch natively exposes a JSON-over-HTTP REST interface. Parsing JSON payloads at massive query rates incurs substantial serialization/deserialization overhead and CPU churn.

• Protobuf and gRPC: Yelp replaced the JSON API with gRPC and Protocol Buffers for all client-to-server and primary-to-replica communications (including segment chunk transfers).
• gRPC Gateway: For legacy internal services that cannot natively communicate over gRPC, Yelp implemented grpc-gateway, an autogenerated reverse-proxy that converts REST/JSON requests into binary gRPC calls.

2. Rapid Node Recovery via AWS EBS Volumes

When deploying search nodes backed by raw Lucene, a standard failure recovery approach involves pulling full index snapshots from object storage (like AWS S3). However, downloading hundreds of gigabytes of segment data over the network creates long bootstrapping delays during failovers.

Yelp engineered a recovery pattern leveraging AWS Elastic Block Store (EBS):

[ Primary Node (EC2) ] ──Writes Index──▶ [ AWS EBS Volume ]
         │ (Crashes)
         ▼
[ Standby/New Node (EC2) ] ──Mounts──▶ [ Same EBS Volume ]
(Ready in seconds)

1. Search nodes store Lucene segment indices directly on attached AWS EBS volumes rather than ephemeral local instance storage.
2. When an EC2 instance hosting an nrtSearch node fails, a replacement instance is provisioned in seconds.
3. The existing EBS volume is detached from the failed instance and attached to the new instance.
4. The new node discovers the pre-existing, intact Lucene index immediately upon mount, avoiding multi-gigabyte network transfers and reducing MTTR (Mean Time to Recovery) from tens of minutes to seconds.

3. Primary-to-Replica Synchronization Flow

With segment-based replication, the lifecycle of a write follows this model:

1. Indexation: The primary node accepts document writes and indexes them locally into memory buffers, flushing them to immutable Lucene segments on disk.
2. Notification: Once a new segment is committed, the primary emits a notification to registered replica nodes.
3. Segment Fetching: Replica nodes pull the new immutable segment files from the primary over a high-throughput gRPC stream.
4. Reader Refresh: Replicas register the newly copied segment into their local searcher pool without executing any tokenization or inverted index computation.

---

Query Performance Optimizations

Virtual Sharding (Segment Slicing)

Running concurrent searches across individual segments can easily cause thread exhaustion if an index contains dozens or hundreds of small segments. Spawning a unique search thread per segment creates high context-switching overhead and memory pressure.

Yelp introduced Virtual Sharding (or segment slicing):

• Segments are greedily sorted based on their document count and size.
• Multiple smaller segments are aggregated into a logical Slice (virtual shard).
• A dedicated search thread is allocated to process an entire slice rather than a single segment.

This greedy binning algorithm guarantees that each search thread receives roughly the same amount of computational work, preventing stragglers and maximizing CPU core utilization.

Parallel Field Fetching

In information retrieval, a query consists of two distinct stages: finding the matching document IDs (DocIDs) and loading the actual stored fields for those documents. nrtSearch parallelizes the stored field retrieval stage across multiple worker threads, significantly speeding up the final response generation for large result pages.

Segment-Level Search Timeouts for Predictable SLAs

In latency-critical applications, a single long-running query on a large, un-merged segment can degrade tail latency (p99/p999).

nrtSearch implements timeouts at the segment evaluation boundary. If evaluating a segment exceeds a preconfigured time budget, that segment aborts execution and yields an empty or partial result set for that partition. The engine consolidates results from all other segments that completed within the deadline.

Trade-off: nrtSearch prioritizes strict latency SLAs over exhaustive recall in tail scenarios, ensuring upstream services never experience catastrophic cascading timeouts.

---

Zero-Downtime Migration Strategy

Migrating Yelp’s primary discovery layer from Elasticsearch to nrtSearch required absolute guarantees around result accuracy and system availability. Yelp managed this migration using their internal Apollo Proxy layer.

                             ┌────────────────────────┐
                             │      Client Query      │
                             └───────────┬────────────┘
                                         │
                                         ▼
                             ┌────────────────────────┐
                             │      Apollo Proxy      │
                             └─────┬────────────┬─────┘
                                   │            │
                   Primary Traffic │            │ Shadow Traffic (5% -> 100%)
                                   ▼            ▼
                         ┌───────────────┐ ┌───────────────┐
                         │ Elasticsearch │ │   nrtSearch   │
                         │ (Production)  │ │ (Validation)  │
                         └───────────────┘ └───────────────┘

Phase 1: Dark Launching (Shadowing)

1. All production traffic routed through Apollo was forwarded to the live Elasticsearch cluster (100% of user responses served from Elasticsearch).
2. Apollo asynchronously mirrored a small fraction (e.g., 5%) of production queries to nrtSearch.
3. The proxy captured responses from both engines and ran automated diffs to verify:
   • Scoring parity and ranking correctness
   • Structural schema alignment
   • Exception rates and boundary conditions
4. As confidence grew, the mirrored query volume was gradually ramped up to 100%.

Phase 2: Phased Production Cutover

Once parity was proven and performance metrics were verified at full shadow traffic:

1. Apollo switched user-facing traffic in controlled increments (e.g., 5% live to nrtSearch, 95% to Elasticsearch).
2. Error budgets, latency percentiles, and business engagement metrics were monitored at each increment.
3. Yelp gradually incremented the live traffic percentage until nrtSearch served 100% of all discovery queries, decommissioning the legacy Elasticsearch clusters safely.

---

Summary of Architectural Trade-offs

Feature / Metric	Elasticsearch	Yelp nrtSearch
Replication Level	Document-level (re-indexes on every replica)	Segment-level (copies immutable Lucene files)
Replica Hardware	High CPU required for tokenization/indexing	Lower CPU; optimized for memory & read I/O
Network Transport	REST / JSON over HTTP	gRPC / Protocol Buffers (with REST Gateway)
Node Recovery	Shard allocation & rebalancing across cluster	Instant EC2 attach of existing AWS EBS volumes
Query Concurrency	Shard-level parallelism	Virtual sharding (greedy segment slicing)
Timeout Boundary	Node/Query level	Segment-level search timeouts

By cutting out general-purpose abstractions and focusing strictly on the mechanics of Lucene’s immutable segments, Yelp transformed an operationally complex, over-provisioned search fleet into a lean, predictable, near-real-time search platform.