Post Syndicated from Netflix Technology Blog original https://netflixtechblog.com/timeseries-tiered-storage-journey-kafka-flink-streams-to-native-cassandra-cold-reads-e59d597c9d60

By Oleksii Tkachuk, Rajiv Shringi, Chris Lohfink, Abe Ratnofsky, Kartik Sathyanarayanan

Introduction

In an earlier post, we introduced the TimeSeries Abstraction, our high-throughput, low-latency platform designed to store and query temporal event data at scale. It has since become the operational backbone for Netflix’s largest datasets, which now span multiple petabytes.

However, managing such large datasets in hot storage, tuned for low-millisecond latency access, is not cheap. At Netflix, cost-efficiency is a continuous discipline rather than a temporary initiative. This post shares two solutions that significantly reduce storage costs by continuously moving older, less frequently accessed data into cheaper object storage:

• Solution 1: Uses Apache Kafka and Flink to stream data directly to S3
• Solution 2: Leverages Cassandra backups in S3 and pluggable Memtables to read data directly from S3

We’ll start with the Kafka/Flink-based design, then describe how we evolved to a native Cassandra-based cold tier.

Challenges

As Netflix’s TimeSeries datasets have expanded, we’ve encountered several compounding challenges:

• Growing retention demands: Users often push for increasing data retention to serve a broader range of use cases, which in turn increases the storage costs.
• Escalating costs: Maintaining massive, dense clusters for infrequently accessed historical data is prohibitively expensive.
• Sparse historical access: Most queries target recent events, but older data must remain readily available for occasional audits and investigations.
• Operational strain: High storage density on hot nodes complicates scaling, deployment, and cluster stability.
• Latency requirements: Even historical data sometimes requires sub-second query performance for internal tooling and analysis.

This creates a clear opportunity: keep the most recent data on fast storage and migrate older, rarely read data to cheaper storage while preserving a consistent API. Done well, this can meet user and operational needs while greatly reducing our infrastructure spend.

TimeSeries Abstraction

Before we dive into the cold storage solutions, here is a quick recap of how TimeSeries partitions its dataset into multiple tables, each covering a window of time we call Time Slices.

This enables several key optimizations, most notably allowing a table to become immutable once it crosses the write boundary, which lets us safely apply further optimizations. For more details, refer to our previous post.

This slice information is critical and is maintained in a dedicated metadata table:

Creating new Time Slices and maintaining their metadata is a primary responsibility of the TimeSeries control plane. The TimeSeries data plane uses this information to correctly route the read and write traffic to the provisioned tables.

With that background, let’s look at the different solutions we built for data tiering.

Solution 1: Streaming with Apache Kafka and Flink

Our first solution followed a familiar streaming pattern using Apache Kafka and Flink, where we streamed data to S3 in realtime while simultaneously persisting it to Cassandra.

Getting Data Into S3 Efficiently

To optimize data streaming to S3, the system hash-partitions Time Series IDs into buckets. This partitions the stream and lets dedicated Flink tasks handle each bucket, ultimately batching and reducing the write throughput to S3. Flink workers batch events into Parquet files grouped by event time within the hour. These files are great for high-throughput streaming writes, but not for low-latency reads.

Enter the Compaction job.

Compaction Job to Optimize Reads

The compaction job runs daily and merges small Parquet files into a single, optimized daily Parquet file per namespace.

The job further optimizes the Parquet files in the following ways:

• Small Row Groups: Creates small (4–8 MB) row groups to minimize lookup cost. We can then use the Parquet Java library to perform range reads of row groups from S3. This is intentionally well below Parquet’s recommended row group floor — smaller groups were better because our reads are dominated by point/range lookups rather than full scans.
• Event Sorting: Sorts events within a row group by time, eliminating brute-force scans when querying for a time range.
• Bloom Filters: Caches Bloom filters and file metadata in EVCache, allowing the TimeSeries query planner to efficiently skip unnecessary reads from S3.

Once the compaction job completes, all compacted files per partition are recorded in the TimeSeries control plane, and the hot storage slices can now be truncated!

On some of our largest datasets, this job typically takes about 12 hours, prioritizing lower compute costs over execution speed.

Streaming Cold Tier Read

Once compacted files are optimized in cold storage, they are ready to be read. The TimeSeries data plane fans out a given query across both the hot and cold tiers and merges their result. The users still use the same query interface:

The Cold Tier Data Plane uses compaction metadata for aggressive planning and pruning, which dramatically reduces the S3 query fan-out. As a result, retrieving a full day of data requires at most three sequential S3 reads, even in the uncommon case where an ID spans two row groups. This yields a p50 latency of ~150 ms and a p99 latency of ~500 ms per read.

At the time of writing, the streaming cold tier implementation was processing up to 1,250 reads per second from cold storage globally:

Why not stop here?

The streaming cold tier has delivered real savings, shrinking both Cassandra’s EBS footprint and the Cassandra backup footprint in S3. But the streaming infrastructure itself is expensive and adds moving pieces that widen the team’s on-call scope:

We wanted to improve this further and asked ourselves:

• Can we avoid streaming data into the object store and the associated data transfer costs?
• Can we reduce the compute and operational overhead of new components?

That led to a more radical question: Can we read directly from existing Cassandra backups in S3?

Solution 2: Cassandra Cold Storage for Immutable Data

At Netflix, we back up Cassandra data into S3 at regular intervals for disaster recovery. The idea for tiering is simple:

Read directly from these S3 backups instead of paying a separate data transfer cost to S3.

Once TimeSeries marks a hot Time Slice as ready for cold storage, we can customize Cassandra to read directly from its backups, allowing us to delete the on-disk data copy. This would also incur minimal changes to the TimeSeries read path!

To make this viable in production, we had to answer a series of “How…?” questions.

How Can We Read the Backups Efficiently?

We introduced a new compaction strategy in Cassandra, which we call ReadOnlyCompactionStrategy (ROCS). Here is how it works:

Once a time slice crosses the user-configured cold-tier boundary, TimeSeries stops accepting writes for that time slice making it immutable. This immutability unlocks significant optimization opportunities:

• SSTable overlap: Standard strategies (LCS, TWCS) often produce SSTables that overlap in the token space, and so a single partition key may exist across multiple files.
• Read fan‑out: Because a read must touch every SSTable covering the key’s token range, this overlap turns a single point-read into multiple lookups, increasing latency across all tiers.

ROCS re-compacts all SSTables once, aligning them precisely with token ranges:

ROCS applies more broadly to all TimeSeries use cases, driving tail latency down on all storage tiers by reducing the SSTable fan-out during read operations:

Once the immutable Time Slices have been optimized, it is time to trigger backups.

How Would We Leverage Backups?

Cassandra backs up every node in the cluster, organizing each backup on S3 as a declarative manifest, along with the SSTables backing the keyspace’s tables:

 {bucket}/
  └── {prefix}/                             ← cluster identity (env_backup/hash_app)
      └── {token}/                          ← node identity (ring token)
          ├── META_V2/                      ← manifests, one per snapshot
          │   ├── {snapshotInstantMs}.json
          │   ├── {snapshotInstantMs}.json
          │   └── ...
          └── SST_V2/                       ← actual SSTable files
              └── {backupTimestamp}/        ← snapshot timestamp
                  └── {keyspace}/
                      └── {table-with-uuid}/ 
                          └── {compression}/
                              └── {encryption}/
                                  ├── {component-file}
                                  └── ...

The backup manifest orchestrates the backup, listing the paths to all its SSTables:

manifest.json
  ├── info
  │   ├── appName              "cass-cluster"
  │   ├── region               "us-east-1"
  │   ├── snapshotInstantMs    1748908800000
  │   ├── backupPathPrefix     "prod_backup/1234_cass/token"
  └── data[]
      └── [0]
          ├── keyspaceName     "ks"
          ├── columnfamily     "tbl-a1b2..."
          └── sstables[]
              └── [0]
                  └── components[]
                      ├── na-1-big-Data.db
                      ├── na-1-big-Index.db
                      ├── na-1-big-Filter.db
                      ├── na-1-big-Statistics.db
                      ├── na-1-big-Summary.db
                      ├── na-1-big-CompressionInfo.db

Alongside the large Data.db file, backups include a compact set of index, metadata, and integrity files. This structure allows us to quickly locate, validate, and restore specific partitions or data ranges without scanning the entire dataset. Furthermore, since the manifest catalogs every S3 path, we can download most of these components directly to the Cassandra node (excluding the Data.db file) where they occupy negligible space. These local components enable efficient query planning before any data is read from S3.

How Would Cassandra Read From Backups?

Cassandra reads typically pull data from two sources: the Memtable, which holds recent unflushed writes, and flushed SSTables on durable storage. As of Cassandra 4.1, we can leverage pluggable Memtables to redirect reads from “cold” tables directly to the backup storage in S3.

-- create table for a given Time Slice
CREATE TABLE data_20260616_00_04 (
    ...
    ...,
    PRIMARY KEY ((time_series_id, time_bucket, event_bucket), event_time, event_id, key)
) WITH CLUSTERING ORDER BY (event_time DESC, event_id DESC, key DESC)
    ...
    AND compaction = ...
    AND compression = ...
    AND memtable = 'backupmemtable:timestamp=1781953200000' ← Custom memtable
    ...

The timestamp marks the point at which the backup is guaranteed to cover the full table, and the table is guaranteed to be immutable.

Opening a backup Memtable kicks off a state-machine initialization that pulls the index data from S3 onto the node once:

This process ensures all SSTable components are downloaded reliably through several resilience mechanisms:

• Retries and backoff: All S3 operations and file-system cache writes use retries with exponential backoff.
• Resource isolation: Initialization runs on a dedicated thread pool, isolated from normal Cassandra operations.
• Partial download protection: If a downloaded component’s size doesn’t match the manifest, it is deleted and re-downloaded.
• DDL change protection: A .timestamp file marks completed initialization and ties it to the backup timestamp. If the table’s backup timestamp changes, this file is removed and initialization restarts for the new timestamp.
• Read protection: Memtable initialization is asynchronous; reads to an initializing memtable block for at most half the read RPC timeout.

Once initialization has pulled every SSTable component (except Data.db) from S3 onto the local disk, the Memtable is ready to serve reads. Here is what the read process looks like:

To minimize latency, the system parallelizes S3 reads across requests, SSTables, and individual chunks. A shared cache deduplicates fetches and prefetches data, so chunks are ready before they are needed.

We further optimize cluster and table settings to reduce the number of S3 calls:

• Larger compression chunks: Reduce S3 calls per partition by reading fewer, larger chunks (accepting some over-fetch on small reads).
• Read repair disabled: Avoid expensive, unnecessary repair traffic against immutable, pre‑repaired backup data.
• Speculative retries disabled: Prevent S3’s higher latency from triggering redundant, overlapping reads.

How Would We Know When It Is Safe To Read From Backups?

However, before the Memtable can point to a backup, we must confirm two things across the whole ring, and we use Cassandra Virtual Tables to do that.

First, that ROCS re-compaction succeeded:

cqlsh> SELECT * FROM netflix_views.cluster_view                                                                                                                                                          
       WHERE keyspace_name = 'system_views' AND table_name = 'sstable_tasks';

keyspace_name | table_name    | host     | value                                                                                                                                                        
--------------+---------------+----------+------------------------------------------------
system_views  | sstable_tasks | 10.0.0.1 | [{"table_name":"<.>","kind":"compaction", ...}]
system_views  | sstable_tasks | 10.0.0.2 | [] <---- no active compactions
system_views  | sstable_tasks | 10.0.0.3 | []

And second, that backups were fully captured (denoted by uploaded=True):

cqlsh> SELECT * FROM netflix_views.complete_backups                                                                                                                                                      
       WHERE keyspace_name = 'my_ks' AND table_name = 'my_table';

keyspace_name | table_name | timestamp     | num_tokens | num_uploaded | uploaded                                                                                                                       
--------------+------------+---------------+------------+--------------+----------
my_ks         | my_table   | 1765211400000 |          3 |            3 |     True
my_ks         | my_table   | 1765213200000 |          3 |            2 |    False

How Would We Automate This Entire Flow?

The TimeSeries control plane automates the entire transition as slices cross the immutable boundary:

Each slice can be in any lifecycle stage at any time without impacting system availability, and the system remains resilient to delayed backups or lagging compactions.

How Would Cassandra Handle The Additional Load?

Serving data from S3 is much more CPU-intensive than reading from local storage. A sudden spike in read requests can overwhelm the Cassandra cluster and affect the hot path, turning a cost-saving feature into a reliability risk.

To avoid this, cold reads can optionally run on a separate Cassandra cluster that points its backup Memtables at the hot cluster’s backups:

The cold Cassandra cluster must match the original cluster in topology but need not match it in hardware! We can run the cold cluster on much smaller, cheaper, storage-light instances.

Real World Performance

Cassandra Cold Tier was rolled out to the most storage‑intensive TimeSeries Abstraction clusters. It currently stores over 15 PB of compressed data in Cassandra cold storage and has allowed us to cut attached storage significantly on all affected clusters:

At the same time, it serves thousands of reads per second from cold tier storage in any given region:

Cassandra Cold Tier improves p90 latency by ~30% over the previous cold storage implementation, thanks to more efficient S3 access patterns and smaller read sizes, while keeping p99 latency under ~500 ms. Cassandra already handles bursts of cold reads gracefully via its chunk cache, so this design preserves that same resilience for the S3-backed cold tier:

Acknowledgements: Special thanks to our stunning colleagues who contributed to the TimeSeries Abstraction’s success and collaborated with us on the cold storage solutions: Kaidan Fullerton, Matt Lehman, Sumanth Pasupuleti, Carlos Matias Herrero, Nick Pavlakis, Bryan Keller, Francesco Di Chiara

---

TimeSeries Tiered Storage Journey: Kafka/Flink Streams to Native Cassandra Cold Reads was originally published in Netflix TechBlog on Medium, where people are continuing the conversation by highlighting and responding to this story.

Post Syndicated from Netflix Technology Blog original https://netflixtechblog.com/behind-the-scenes-evolving-netflixs-ads-event-pipeline-for-live-part-ii-826ebf9ad9fb

By Yogesh Nagarur

Introduction

In Part 1 of this series, we shared how Netflix built the Ads Event pipeline and approached it as a system design problem: replacing fragmented pipelines with a centralized collection, enrichment, and a standard data contract. The result was Ads Event Publisher, the system that collects ad telemetry from client devices, enriches it with the context of the ad that was actually served, and publishes a single, unified stream of ad events to every downstream consumer: frequency capping, metrics, billing, measurement partners, and reporting. Any member viewing or account details that move through this pipeline end-to-end are encrypted and handled in accordance with Netflix’s strict privacy and security practices to safeguard member PII.

If ad serving is the “brain” and ad events are the “heartbeats” that feed it, then enrichment is what gives each heartbeat its meaning, turning a bare “[Somebody] watched an ad to completion” signal into “this campaign, this creative, at this price, for this break.” Part 1 described that enrichment step as a single box in the architecture. This post is about what was inside that box, and why we rebuilt it.

Part 1 ended with a look at what was on the horizon: ads on live streams, de-duplication, and richer data signals. A single project delivered all three by rebuilding the Ads Event Publisher around stateful stream processing. We replaced a database lookup with a real-time streaming join in Apache Flink, which let the system scale for Live, took a dependency off the critical path, and replaced the engine of a high-traffic pipeline without downtime.

Let’s dive in.

A quick recap: enrichment and the Metadata Registry

When an ad is served, the ad server knows everything about it: the campaign, the creative, the price of the impression, the targeting context, and the tracking URLs that measurement partners expect us to fire. The client device, by contrast, knows almost nothing: just an opaque token and when to send each event. Enrichment is the act of reuniting the two.

In the architecture from Part 1, that reunion happened through the Ads Metadata Registry, a Key-Value store:

• When the ad server made a decision, it wrote the metadata for that impression into the registry.
• It encoded only reference identifiers into the opaque token returned to the device, keeping the token small.
• Later, when the client fired an event carrying that token, the Ads Event Publisher decoded the token, looked up the metadata in the registry, and stitched the two together into an enriched ad event.

This design served us well. The registry was globally replicated, which gave us a useful property almost for free: an event arriving in any region could find its metadata regardless of where the ad was decided. It future-proofed the token against ever-growing metadata, and it carried us through the first phase of Netflix’s advertising growth.

But as we looked ahead, especially to ads on live streams, the limits of this design became clear.

Why we needed to evolve

The forcing function was Live. Netflix’s live slate has grown rapidly, from NFL games to blockbuster fight nights to the FIFA Women’s World Cup 2027, and viewership keeps growing. That traffic looks nothing like Subscription Video on Demand (SVOD): a single event can surge toward tens of millions of concurrent viewers within minutes, then drain just as fast when it ends. Building the ad pipeline for Live meant building for that trajectory.

To see why the old design could not get there, let’s put some numbers on it. When Netflix streamed the Jake Paul vs. Mike Tyson fight to more than 100 million viewers, a single ad break at that scale would generate an enormous volume of ad events. Serving those ads in the old design would have meant writing metadata for each of them to the Metadata Registry and waiting for acknowledgment before the ads could serve: millions of must-succeed database writes per second on the ad-serving critical path. That pattern created three core problems at Live scale:

1. It was a scaling bottleneck. Every ad and every ad break is a write. Multiply that by tens of millions of concurrent viewers, keep it non-lossy (a dropped write is a dropped ad, and lost revenue), add Live’s spiky on/off pattern, and the write load becomes enormous. But you cannot safely scale the registry up once the event starts. Adding nodes reshuffles data across the cluster to rebalance it, and doing that in the middle of a live event could slow or even fail the very writes you are adding capacity for. That leaves one safe option: pre-scaling for the peak in advance and hoping the forecast is accurate.
2. Fewer dependencies on the critical path. Serving an ad meant a synchronous, must-succeed write to the registry, followed by a wait for its acknowledgment. Every system on the critical path is one more thing that can stop an ad from serving, and the metadata registry had no fallback, so an unhealthy registry meant serving stopped. Taking hard dependencies off that path was the goal, and the registry was the biggest one we could remove.
3. It put event freshness at risk. If writing slowed down under load, or lookups timed out, events spilled into retries and dead-letter queues, and enriched ad events were delivered late. Some systems might be fine, but others, such as frequency capping, cannot. It needs events under a strict SLA to decide whether the next selected ad can run without breaking policies.

So we asked a simple question: the ad server already knows and logs everything about the ad it served. Does it really need to write that to a database on the hot path? Can we match that against the downstream Kafka stream of client events, which is always there, after the fact?

The write leaves the critical path entirely; no shared datastore outage can stop serving, and enrichment becomes something we scale elastically with Live rather than a cluster we babysit.

Matching two streams after the fact is a streaming join, and that is where Flink comes in.

Why Flink

We chose Apache Flink because near-real-time, stateful stream processing is exactly what it is built for, and because it has deep, production-tested platform support at Netflix. Just as important, we already knew how to run it. As Part 1 noted, our Ads Sessionizer is a large stateful Flink job, so operating one at scale, tuning state, checkpoints, and recovery under production load, was familiar ground for the team. Enrichment on Flink was built on that experience rather than starting from scratch.

The new architecture

At its core, the new Ads Event Publisher is built as an Apache Flink streaming job. Each event moves through a handful of stages:

1. Normalize. Client events arrive from several sources, including SVOD, live, playback, and embedded ad breaks, each with a slightly different shape. We first normalize them into a common form.
2. Re-route. Client events and response events may appear in different regions; we inspect and reroute them to the correct job-processing region for a successful join.
3. Join. The heart of the job: we match each client event to the ad-serving response for the same impression, enriching it with the full serving context using a KeyedCoProcessFunction
4. Deduplicate. Revenue-bearing events are de-duplicated, so nothing is counted twice.
5. Publish. We extract measurement beacons for our partners and route the enriched ad event to its downstream consumers.

The join is the interesting part. Client events and ad serving responses arrive on separate streams at different times, so the job holds recent responses in memory and matches events against them over a 60-minute window. Three things can happen:

• The response is already there when an event arrives → we enrich it on the spot.
• The event arrives first → we hold it for a configuration duration (currently set to 60 minutes) until the matching response shows up, then enrich it.
• No response arrives within the window → we mark the event as unmatched for later recovery.

We built the join as a custom Flink operator (a KeyedCoProcessFunction) rather than a built-in windowed join. We do not need a windowed join ordering and time-window semantics, since downstream consumers handle ordering; we simply match each client event to its response by key. Owning the operator at its lowest API form lets us control the matching state directly, including how long we hold it and what we do with an event that never matches. This allows us to tune retention per stream or persist unmatched events to a durable store later without rewriting the join. Our goal is always to build a strong base to reap the benefits later as we scale.

To make that concrete, picture a member watching a live game. When the ad server decides which ad to show, it logs its response, and a few seconds later, the member’s device fires a “first quartile” event carrying the opaque token. Both arrive at the job keyed to the same impression, and because the response is already waiting, the event is enriched on the spot with its campaign, creative, price, and tracking URLs, then published downstream in near real-time.

Why 60 minutes? Almost all events, roughly 99.999% of them, reach the job well before that, so extending the window further would only grow the state store and lengthen checkpoints for a vanishingly small gain. The remaining 0.001% still matters for precise billing and reporting, and we recover those events via a separate path described later on.

Cross-region: matching events where their data lives

The Metadata Registry handled cross-region enrichment for us because it was built on a globally replicated key-value store: a client event could land in any region and still find its ad’s metadata. Global replication was a feature the platform provided (see Netflix’s Key-Value Data Abstraction Layer), so serving an ad in one region and enriching its events in another was simply not something we had to think about.

A streaming join does not come with that convenience, and here the two sides of the join live in different places. The ad serving response stays in the region where the ad was served. The client event, though, is fired by the device, and the device can be anywhere: a member might start an ad on home wifi and switch to mobile data partway through, and that network change can route the event to a different region than the one that served the ad. So a client event is routed to a region that does not have the matching serving response.

The tempting fix is to stream every region’s serving responses into every other region so that each region can match any client event locally. We rejected it: that would mean every region storing every serving response worldwide, the vast majority of which will never be used in the join, which is wasteful in both cost and compute.

Instead, we treat this as a routing problem and route the client event rather than replicating the response. We encode the serving region into the ad token, so when a client event arrives in the wrong region, we forward it to the region that owns its response. For example, an ad is served in us-east-2, the member’s device switches from wifi to mobile data mid-ad, and the resulting event arrives in us-east-1. That region reads the token, sees the ad belongs to eu-west-1, and forwards the event back there, where its serving response is waiting. Only the small trickle of genuinely cross-region events pays for the extra hop.

Catching the long tail of late-arriving events

Sixty minutes covers the vast majority of events, but not all of them. A device may have been offline, an ad paused, or an engagement event delayed by hours. Rather than drop these stragglers, we set them aside and recover them through a companion path: an hourly Spark batch job, orchestrated by Maestro, Netflix’s workflow orchestrator, re-pairs each straggler with its serving response from our data warehouse (which has a far longer retention than our in-stream window) and publishes it through the same downstream sinks. This keeps the streaming job’s memory bounded while still catching the long tail.

This also lets us support event types that arrive well after the 60-minute window, including certain tracking-related events, while keeping the door open to extending the system to additional event types down the line and improving our match rate as needed. We also designed this offline job with reuse in mind: it can replay dead-letter queues and backfill historical data, giving us a single mechanism for both long-tail recovery and operational recovery.

What was hard

Beyond the architecture itself, three things took real work, and together they are what made the pipeline stable, correct, and safe to switch on.

Keeping the core lean.

Operating a stateful streaming join at live-event scale took real tuning, and most of our hard-won lessons come back to one idea: the job’s only job is matching events. Anything else we were tempted to bolt on, such as external lookups or incidental side-processing, was better moved off the hot path, or it eventually showed up as instability under load. Getting this right was the difference between a job that comfortably kept pace with live traffic and one that fell behind. It is unglamorous work, and it is most of the work.

De-duplicating Events

Duplicates are a fact of life here in an event processing pipeline. A client may retry an event, and our upstream delivery is at-least-once so that the same event can arrive more than once. For revenue-bearing events, the ones that feed billing, nothing should be counted twice.

The solution here is a simple and idempotent key. For each event, we compute a stable ID by hashing a few fields that uniquely identify it, such as the ad ID, the event type, and, for repeatable events, an offset or timestamp. The same underlying event always hashes to the same ID, so if we have seen an ID before, we drop the repeat.

What changed with Flink is where we remember those IDs. The previous design kept them in an external key-value store and did a check-and-write over the network for every event; Flink lets us keep that state local to the job instead. We key the stream by event ID, which routes every copy of a given event to the same place, and a local state store with a short time-to-live remembers the IDs we have recently seen. Keying by ID also removes the race the external store had to guard against, since a given ID is handled in only one place at a time, with no network hop on the hot path. Multiple improvements here alone.

We are deliberate about the guarantee: at-least-once delivery with best-effort de-duplication, not exactly-once. In-stream dedup removes the vast majority of duplicates within its window. However, a few can still slip past, for instance, through a cross-region reroute, a late-arriving recovery, or a restart that replays recent events. So we also carry the event ID through to the enriched ad event itself, which lets consumers who need exact counts, such as billing, run their own idempotent de-duplication over their own, longer windows.

Production Rollout

This was the most challenging part of the project. For a system that feeds billing and frequency capping, “trust me, the new one is correct” is not good enough, so we ran the Flink pipeline alongside the legacy one and made it earn its traffic.

Both pipelines continuously processed the full event stream, allowing us to compare them in real time: we published their outputs side by side. We ran automated audits that alerted us when the two drifted apart. Those audits earned their keep, surfacing subtle discrepancies we would never have caught from dashboards alone.

How do we know the enrichment logic between the two systems was equal for the audits to make sense? Early in development, we extracted that logic into a shared module that both the legacy service and the Flink job depend on, so any change to how an event is enriched is applied to both at the same time. A failure in the audits, therefore, pointed to a real difference in the pipeline mechanics.

Cutover was a separate question from comparison, and this is the part that warrants precision. We could not simply send a random percentage of events to each pipeline, because the same event landing in both would be double-counted downstream. Instead, the split was deterministic: a stable key on each event decided which pipeline was responsible for publishing it, so every event was published exactly once, even though both pipelines saw all of them. Raising the threshold moved events from the legacy pipeline to Flink a slice at a time, and we could dial it back instantly if anything looked off.

By the time we fully cut over, the two pipelines agreed on well over 99.99% of events, and we rolled it out region by region rather than all at once. Then came the satisfying part: we turned off the ad server’s writes to the Metadata Registry, and decommissioned the legacy service and the registry itself. The single point of failure on the ad-serving critical path is simply gone. One of the oldest ad serving components at Netflix was deleted.

Operating the Flink platform

Operating a large stateful streaming job in production is its own job, and we did not build that muscle alone. We leaned heavily on Netflix’s Flink platform team. Early on, the job’s checkpoints were slow and heavy. The first checkpoint in particular took a long time, and we were holding more state in memory than we needed. Working with the platform team, we trimmed the state, reduced checkpoint durations, and validated the gains in load tests. That is what keeps the job from stalling during Live bursts.

The same team also gave us operational scaffolding we did not have to invest in, including continuous delivery and canary deployments, so we could ship and roll out the Flink job with the same guardrails and confidence as the rest of our production systems. Netflix operates more than 25,000 Flink jobs (see Building a Scalable Flink Platform at Netflix), and that maturity is what we have to build on. We layered on custom canary metrics that catch issues before a release, CI/CD that lets us stagger rollouts, and a deploy flow that keeps the old job running until the new one has detected healthy checkpoints, so we never destroy a working job before the new job gives us the green light.

The results

• Enrichment is off the critical path for ad serving. The ad server no longer performs a synchronous, must-succeed database write to serve an ad. It logs and moves on, and an entire class of “the lookup store is down, so ad serving is down” failures no longer exists.
• Fewer systems, fewer schemas, fewer hops. We retired both the legacy service and the Metadata Registry.
• Built for Live, and continuously tested for it. An ad break is what this service is designed to handle, so during our journey, we also ran large-scale load tests to simulate real-life traffic. In our latest load test, a single AWS region handled one ad break for a large concurrent audience, processing hundreds of thousands of events per second, joining the large majority of them against their serving responses in real time, and outputting enriched ad events at a comparable rate, all with negligible timeouts and errors. We test this throughout development and keep doing so as live audiences grow, and we are not at the ceiling yet.
• We delivered the Part 1 roadmap. Live-stream ad events, in-stream de-duplication, and richer real-time enrichment all shipped on this foundation.

Key takeaways

• Take the database off the hot path when you can. Our biggest win wasn’t raw performance — it was blast radius. Trading a synchronous write-and-wait for a fire-and-forget log plus a downstream join removed an entire failure mode from ad serving.
• Match events where their data lives. Real-time joins beat global replication by routing events to their data rather than copying the data everywhere.
• Earn the cutover. Running both systems in parallel and diffing their output on live traffic was slower than a flag-day switch — and it’s the only reason we trust the result.

What’s next

We have an exciting list of projects on the horizon: scaling the live path for the next tier of concurrent audiences, bringing the remaining delayed engagement events fully into the streaming world, reducing the Flink state size we need to manage, working on bringing even more resilience to the eventing processing pipeline, and continuing to add new event types and richer signals for our downstream consumers.

Re-architecting the engine of a running, revenue-critical pipeline in place was a real multi-team effort. If building real-time systems at this kind of scale sounds like your thing, Ads Engineering is hiring.

Acknowledgements

This was a multi-team effort across Netflix. Thank you to everyone who designed, built, validated, and operated the new pipeline:

Ads Engineering: Yogesh Nagarur, Kinesh Satiya, Jeet Mehta, Nitin Saini, Persis Shah, Guangyu Chen, Jimmy Liu, Riley Berton, Simon Spencer, Aadesh Patel

Ads Data Engineering: Eugene Yusim, Brandon Nicklas, Indu Narayan

Ads SRE: Max Michaels, Kevin Chung, Carlos Avalos, Brian Christie

Data Movement: Mark Cho, Herbert Wang, Shraddha Ladda, Nikhil Sulegaon, James Lubin, Daniel Trager

---

Behind the Scenes: Evolving Netflix’s Ads Event Pipeline for Live — Part II was originally published in Netflix TechBlog on Medium, where people are continuing the conversation by highlighting and responding to this story.

Post Syndicated from Netflix Technology Blog original https://netflixtechblog.com/how-and-why-netflix-built-a-real-time-distributed-graph-part-3-querying-the-graph-with-grpc-0f3468349607

How and Why Netflix Built a Real-Time Distributed Graph: Part 3 — Querying the graph with gRPC execution API

Authors: Nilesh Mishra and Ajit Koti

This is the third entry of a multi-part blog series describing how we built a Real-Time Distributed Graph (RDG). In Part 1, we discussed the motivation for creating the RDG and the architecture of the data processing pipeline that populates it. In Part 2, we discussed how we designed the storage layer to handle billions of nodes and edges while maintaining single-digit-millisecond latency. In Part 3, we will explore how we designed a fast, flexible serving layer to efficiently query the graph.

Introduction

In Part 1 of this series, we described why Netflix needed a Real-Time Distributed Graph (RDG) and how we used Apache Flink to build an ingestion and processing pipeline that turns streaming events into graph primitives. In Part 2, we explored how we designed a storage layer capable of handling billions of nodes and edges while still delivering single-digit-millisecond latency.

In this post, we focus on the next challenge: querying the graph efficiently to power real-time insights for our internal partners. All of the work on ingestion and storage only matters if we can actually ask complex questions and get answers back quickly. As we optimized for lower latency, we found that the serving layer posed its own set of challenges, distinct from those of ingestion and storage. How do we turn a constantly evolving, billion-edge graph into sub-100ms responses across a wide variety of workloads? This is the problem we tackle in this post.

The Real World Needs

As we integrated the RDG into Netflix’s ecosystem, we realized that “querying the graph” is not a one-size-fits-all operation. We needed to handle a wide range of access patterns: from high-volume security lookups to deep, exploratory personalization traces.

Let’s revisit our example from Part 1 and expand on it slightly. In the earlier posts, we focused on accounts, devices and content. In practice, the graph is richer: each account has multiple profiles.

A member journey often looks like this:

1. Alex logs in to their Netflix profile on a smartphone and starts watching Stranger Things.
2. They later switch to a smart TV in the living room to continue the episode.
3. The next morning, they use a tablet to play the game Stranger Things: 1984.

In the RDG, this journey creates the following graph structure:

Graph queries vary along two axes: how wide they fan out at each hop, and how deep they chain across hops. To see this range, let’s look at two scenarios from opposite ends:

1. Shallow and wide: “Which devices has this account used?”

Consider a “shallow, wide” query: “Which devices has this account used to stream in the last 30 days?”

Using the graph structure above, this translates to:

• Starting Point: A specific Account Node.
• Hop 1 Edge Traversed: The streamed_from edge.
• Hop 1 Destination: Device Nodes.

While this is only a “single hop,” it presents a significant scaling challenge. For a highly active account, the fan-out can be massive. The query layer must fetch hundreds of streamed_from edges, apply temporal filters on each edge’s last_watch_timestamp property to capture only those within the last 30 days, and aggregate the results, all while maintaining sub-100ms latency.

2. Deep & Narrow: What has this profile watched?

Consider a scenario where personalization teams need to understand a member’s viewing journey. They might ask: “For Account X, show me the Stranger Things viewing history across all profiles: which profiles watched it, what they watched, and when”.

This path unfolds as follows:

• Starting Point: A specific Account Node.
• Hop 1 Edge Traversed: has_profile
• Hop 1 Destination: Profile Nodes
• Hop 2 Edge Traversed: started_watching (filtered for title_name = “Stranger Things”)
• Hop 2 Destination: Content Nodes

The core challenge in this scenario is sequential dependency: we cannot fetch a profile’s viewing history until Hop 1 has identified which profiles exist. In a distributed environment, the client has to wait for Hop 1 to finish before sending Hop 2. If each hop takes 10ms of network time, that’s 20ms of overhead before we’ve processed a single byte. To hit our sub-100ms goal, we needed a way to package this multi-step logic into a single request.

This example is a 2-hop traversal, but queries can chain 3–4 hops across different entity types, and the latency penalty of sequential execution only grows with depth.

Balancing Depth and Breadth

These two scenarios pull the system in opposite directions. Shallow-wide queries stress I/O throughput: can we handle massive fan-out without slowing down? Deep-narrow queries stress execution efficiency: can we chain multiple hops without the network overhead adding up? Supporting both on the same system is what shaped the design that follows.

Design Constraints and Key Choices

The two scenarios above sit at opposite ends of the spectrum, but they are not unusual. In practice, the RDG serves tens of thousands of queries per second, each potentially different, all needing sub-100ms responses while the underlying graph continues to grow. Scale, latency, query diversity, and the need for extensibility pulled the design in different directions at once, and every choice came with a trade-off we had to live with.

Why breadth-first, not depth-first? The most intuitive way to traverse a graph is depth-first: pick a path, follow it to the end, backtrack, try another path. But in a distributed system where every hop is a network call, depth-first can lead to high latency. If Account X has 5 profiles and each profile has watched hundreds of titles, depth-first would trace all of one profile’s watched titles before moving to the next, missing the opportunity to batch lookups across profiles. Breadth-first flips this by working one level at a time across all nodes, rather than one path at a time through each node. We fetch all profiles for the account at once, then fetch the started_watching edges for all profiles, and finally fetch content details for all matching titles. Three rounds of parallel calls instead of sequential chains. With breadth-first, there is a clear trade-off in memory, because we hold each level of the graph in memory at once, so the cost scales with how wide a level fans out rather than how deep the query goes. We keep this comfortable by bounding each hop with the per-edge-type limits described in Step 5 below, so even a high fan-out level stays a manageable frontier. We’ll walk through how this works, level by level, in Step 3 below.

Why async-first, not thread-per-request? Latency in the RDG is dominated by I/O, reading from the storage layer, calling enrichment services, and waiting on caches. A traditional thread-per-request model would pin a thread to each in-flight query, and most of the time, the thread would be idle, waiting for a network response. With thousands of concurrent queries, we’d need thousands of threads, most of which would be doing nothing. Instead, we decided to build the entire execution pipeline around asynchronous composition. A small set of dedicated thread pools (16–24 threads total) handles thousands of concurrent requests because no thread ever blocks on I/O. While a storage call is in flight, the thread continues with other work and picks up the result when it arrives. This is the foundational design decision on which everything else rests. We’ll see this in action in Step 4 below, where we cover parallel execution.

Why cache selectively, not everything? Not all data in the graph changes at the same rate. Some properties, such as account plan type and content metadata, are relatively stable: they change on the order of hours or days. Edges like who watched what and when change constantly. For stable data that many queries touch, we use a distributed cache (EVCache) with TTLs tuned to data volatility. Getting the caching strategy right took iteration. We started by caching aggressively and measured the impact: tracking hit rates, monitoring stale-data incidents, and adjusting TTLs based on how quickly different node types actually changed in production. The result: 70–80% hit rates on node lookups, achieved by narrowing the cache to nodes that are both frequently accessed and slow to change, while skipping data that would expire before the TTL ran out. Step 6 below covers how this works in practice.

Why opt-in enrichments, not automatic? Clients know what they need. A query checking account relationships doesn’t care about title artwork; a personalization service building a viewing timeline does. Rather than fetching metadata from external services by default and penalizing every query, we make enrichments opt-in: clients specify exactly which external data they want per request. Also, enrichment is fail-open: if a service is slow or unavailable, we return the graph data without it.

Why eventual consistency, not strong? Most of our queries ask “What has this member done recently?”, not “What happened in the last millisecond?” By defaulting to eventual consistency, we read from the nearest replica and avoid coordination overhead. While the RDG is used to power in-the-moment experiences, it is not set up as the source of truth for the data it holds.

Architecture Overview

The above choices lead to the following three-layer architecture:

The Graph Query Service is the entry point. It accepts gRPC requests, validates the traversal specification, and hands it to the query execution engine. The execution engine orchestrates breadth-first traversal: expanding one level at a time, applying filters and limits at each hop, and composing all I/O asynchronously.

The Storage Abstraction Layer sits between the execution engine and the underlying KVDAL storage. It provides a clean interface for node lookups and edge retrieval, handles streaming for large adjacency lists, and manages node caching (EVCache).

The Enrichment Layer fetches additional metadata from external Netflix services on demand. It batches requests, runs them in parallel with graph data assembly, and degrades gracefully when an enrichment source is unavailable.

When a client sends a query, the request flows through these layers in sequence: the Query Service parses the request into an execution plan, the execution engine walks the graph level by level through the Storage Abstraction Layer, and if enrichments are requested, the Enrichment Layer fetches and merges external data before the response is serialized back to the client.

Now, with that mental model in place, let’s follow a query through this system and see how these choices play out in practice.

Executing Queries Efficiently: Following a Query’s Journey

To see how the RDG query layer works in practice, let’s follow a single query end-to-end and focus on one question: how do we make every step fast?

We’ll reuse the deep-narrow example from above:

For Account X, show me the Stranger Things viewing history across all profiles: which profiles watched it, what they watched, and when.

In graph terms, this becomes a 2‑hop traversal:

1. Account X → has_profile → Profiles
2. Profiles → started_watching → Content (filtered for “Stranger Things”)

We’ll walk through how this query moves through the layers we described above:

1. Reading and interpreting the request
2. Reading from storage efficiently
3. Executing traversal with breadth‑first levels
4. Running many operations in parallel, but safely
5. Filtering smartly to keep only what matters
6. Making repeat queries faster with caching

By the end, we’ll see how a 2-hop query like our Stranger Things example, with streaming, filtering, and parallel execution, can complete in under 100ms.

Step 1: Reading the Request: Deciding What the Query Really Wants

Every query starts as a gRPC request. Before we touch storage or walk a single edge, the engine needs to understand what the caller actually wants.

For our running example below:

For Account X, show me the Stranger Things viewing history across all profiles

The engine creates a traversal plan with a set of levers: how many hops, how many edges per hop, how much history to consider, and whether to favor recent activity.

We resolve these upfront by merging a hierarchy of filters and limits, from application-level defaults down to per-edge-type overrides, into a concrete execution plan. By the time we read from storage, every hop has clear rules. We’ll see how this hierarchy works in detail in Step 5, but the key insight is simple: interpreting the request up front prevents over-fetching from the downstream storage layer.

Step 2: Reading from Storage: Direct Lookups and Streaming Fan‑Out

Once we’ve parsed the request and decided what the query should do, the next step is to actually touch the graph. For our running example:

For Account X, show me the Stranger Things viewing history across all profiles…

The first concrete question the engine has to answer is very simple:

Which profiles does Account X have?

Under the covers, that really means: how do we find all relevant edges for Account X without scanning the entire graph every time?

Finding Edges Fast with Adjacency Lists

If we stored every edge in one massive table, the naive approach would be to scan for rows where source = Account X. Even with indexing, doing that across billions of edges for every request would be slow.

Instead, we organize edges as adjacency lists. For each node, we keep a compact list of “who it’s connected to” by edge type. For Account X, a simplified view might look like:

Account_X: has_profile → [Profile_Alex, Profile_Kids, …,]

Now “get all profiles for Account X” is no longer a global search; it’s a direct lookup into Account X’s stored adjacency. The storage layer can usually pull that list back in a few milliseconds because it’s reading a small, well‑indexed slice of data instead of hunting through everything.

For our query, the first hop is quick: Account X has just two profiles. The engine fetches those edges with has_profile and moves on. For more information on Storage, refer to our previous post.

When One Node Has A Lot of Neighbors

The first hop was small, but the second is where things get interesting. Each profile can have a large number of started_watchingedges. Loading the entire adjacency list at once would spike latency and memory usage.

To avoid this, we treat adjacency lists as streams rather than blobs.

When the engine requests Profile_Alex’s started_watching edges, the storage layer streams them in batches of 100. As each batch arrives, we apply filters (e.g., “last 30 days”) and decide whether to continue.

If we’ve collected enough edges to satisfy the query’s limits ( max_edge_cnt, lookback window, etc.), we stop reading. Otherwise, we pull the next batch.

In our Stranger Things example:

• Storage streams the started_watching adjacency for Profile_Alex.
• There are about 500 edges total: months of viewing history
• As each batch arrives, we filter for Stranger Things and drop anything older than 30 days.
• After a few batches, we’ve found what we need: a handful of Stranger Things sessions.
• We never materialize more data than needed. Filtering happens at the source.

Why This Matters Later

These two choices, the adjacency‑list lookups and streaming fan‑out, enable everything that follows:

• Small fan‑outs (like Account → Profiles) yield predictable, low‑millisecond lookups.
• Large fan‑outs (like Profile → Content) stay efficient by reading only what’s needed.
• Traversal logic treats “neighbors of this node” as a cheap, bounded operation.

By Step 3, we’re working with concise frontiers like “Profile_Alex and Profile_Kids,” ready for the next hop into their viewing histories.

Step 3: Traversal Execution: Walking the Graph Level by Level

We’ve completed the first hop. From Account X, we pulled the has_profile edges and found two profiles: Profile_Alex and Profile_Kids.

But we’re not done. The query was:

For Account X, show me the Stranger Things viewing history across all profiles: which profiles watched it, what they watched, and when

So we still need to fetch each profile’s history and filter it down to Stranger Things sessions. As we covered in our design choices, we use breadth-first traversal: expanding all nodes at the current level in parallel before moving to the next.

Querying, Level by Level

Let’s walk through the Stranger Things query level by level.

Level 1: Account → Profiles

Starting at Account X, the engine pulls has_profile edges, discovering two profiles:

• Profile_Alex, Profile_Kids

These become the frontier for Level 2, a single small lookup that takes a few milliseconds.

Level 2: Profiles → Content (Stranger Things)

From those two profiles, we fetch started_watchingedges and filter for Stranger Things. Instead of exhausting Profile_Alex’s entire viewing history before touching Profile_Kids, we treat this as one logical step:

• For each profile, fetch started_watching edges in parallel.
• Filter for title_name = “Stranger Things” as edges stream in.
• Each profile might have hundreds of content edges, but filtering at the source keeps the result set small.

We discover that Profile_Alex watched Season 1 and Season 2, while Profile_Kids watched Season 4. Level 2 turns “2 profiles” into “a handful of Stranger Things sessions” in roughly one storage round trip.

The traversal completes: two levels, two frontiers.

Why This Matters for Latency

We parallelize within each phase, then regroup. This provides:

1. Predictable resource usage: known requests per level
2. Maximum parallelism: all frontier nodes processed together
3. Far fewer round trips: one per level, not per path

For a 2-hop query: two rounds of parallel lookups instead of hundreds of sequential ones. That’s why our Stranger Things query completes in under 100ms.

Step 4: Parallel Execution: Doing Many Things at Once, Safely

Breadth-first traversal enables parallel work at each level, which is the key to low latency.

At Level 2 of our Stranger Things query, we fetch started_watching edges for each profile. With two profiles, this is trivial, but in production queries fan out across many profiles, each with hundreds of edges to stream and filter. So do we process them sequentially or in parallel? Sequential means waiting for each profile before starting the next, and the delays stack up. Parallel finishes in the time of the single slowest profile, but hundreds of queries doing this at once could overwhelm storage with unbounded concurrency.

The goal: parallel speed without unbounded chaos.

A Kitchen, Not a Single Queue

We structured the query engine like a professional kitchen, with specialized stations for appetizers, mains, and desserts, each with its own capacity. If one station is slammed, the others keep flowing. In practice, that means dedicated thread pools for different work types: fetching nodes, reading adjacency lists, and performing enrichments. When the Stranger Things query reaches Level 2, calls route to the adjacency-list pool, where 8 workers stream and filter each profile’s edges in parallel.

Knowing When to Back Off

Thread pools give us local control, but we also need a global view of total capacity, so we use adaptive concurrency limiting. When things are healthy, we raise the limit gradually (100 in-flight, then 101, 102, and so on); when timeouts or errors spike, we back off by a larger step (say, 100 down to 70). Combined with per-pool limits, the engine constantly tunes parallelism, fanning out within each level while staying inside safe storage and network limits.

Fetching Extra Metadata Along the Way

If the client opted into enrichments (say, maturity ratings for the matched content), the Enrichment Layer fetches them in parallel on its own thread pool and merges them into the response. Enrichment is fail-open: a slow or unavailable source never blocks the query, and we just return the graph data without it.

​​Step 5: Smart Filtering: Keeping Only What Matters

We’ve traversed from Account X to profiles, then to their viewing histories. But raw edges aren’t what our partners need. They care about recent, relevant activity, not every started_watching edge accumulated over the years. This is where filtering decides which parts of the story make the final cut.

From “All Activity” to “The Last 30 Days”

Go back to the original question:

For Account X, show me the Stranger Things viewing history across all profiles: which profiles watched it, what they watched, and when.

The phrase “viewing history” is deceptively simple. Under the hood, it means we need to:

• Ignore older viewing activity, even if it exists in the graph
• Avoid pulling more edges than we actually need
• Let different teams choose their version of “recent enough.”

We handle this with a filtering hierarchy. The system starts with conservative defaults (e.g., 100-day lookback, 300 edges per hop), and requests can override them globally, per-hop, or down to specific edge types. In our query, the 100-day default applies broadly, but the caller sets 30 days for started_watching edges, and the narrower rule wins. Older sessions are discarded. The same engine can just as easily provide a tight recent window on one edge type and full history on another, all in a single query.

Choosing Which Edges to Keep: LATEST vs ANY

Sometimes there are still more edges than we want to return after time filtering. If Profile_Alex watched the same episode several times last month, pausing and resuming, we don’t want to send all those edges back. So we offer two selection modes.

LATEST sorts edges by timestamp and keeps the newest ones up to the limit, ideal for “what has this profile watched recently?” where teams want the current state, not every play event. ANY grabs whichever edges it encounters first, no sorting, which is faster and fine for “has this profile ever watched Stranger Things?” where timing doesn’t matter. Teams default to LATEST and switch specific edge types to ANY when “any proof” is enough.

Bringing It Back to Our Story

So what happens for our running query?

We start with all the started_watching edges for each profile. The time filter narrows this to 30 days. Edge-count limits prevent response flooding. LATEST mode selects the most recent viewing session per title. The result: a concise answer distilled from a verbose history:

• Profiles that watched Stranger Things in the last month.
• Which seasons and episodes they watched.
• The most recent session for each, tying it all together.

This filtering turns raw history into a focused answer.

Step 6: Making It Even Faster: Caching the Things We Keep Seeing

By now, we’ve walked the full path of our query: we’ve traversed from account to profiles, filtered viewing history by time, and focused on Stranger Things sessions.

Despite our optimizations, each storage call still costs a network round-trip. When the same nodes appear across thousands of queries per minute, those redundant calls add up: both in infrastructure cost and in tail latency at scale.

The key question: what can we avoid repeating?

The Things That Don’t Change Every Second

Look back at the entities in our Stranger Things journey:

• The Account node (plan type, region, etc.)
• The Profile nodes (“Alex”, “Kids”, whether it’s a kids profile)
• The Content nodes (Stranger Things seasons and episodes)

These rarely change. Profiles don’t flip between “kids” and “non-kids” every minute. Title metadata is stable.

To improve efficiency, we keep a distributed cache of hot nodes (accounts, profiles, content) that are likely to reappear. When the same entity appears again, we answer “What is this node?” from memory, skipping storage.

Result: for high-traffic entities, we eliminate storage calls and noticeably reduce infrastructure cost and tail latency at scale.

A Quick Replay of Our Query With Caching Turned On

The first time the Stranger Things query runs for Account X, the cache is cold, so we pay the full cost: we fetch the account and its profiles, then the started_watching edges and matching content nodes, caching each node as we go. Minutes later, a different query arrives:

Show me everything Account X’s profiles have watched in the last 7 days, and flag anything rated TV-MA on the kids profile.

This time, many of those nodes are already in the distributed cache. Storage still handles the adjacency lists and edges, but node lookups are lighter and latency drops. At scale, that reuse gives us comfortable headroom for traffic spikes.

Not Everything Deserves a Spot in Cache

We can’t cache everything. The RDG prunes old activity after a set retention window, so caching a node that’s about to be deleted is wasteful.

To avoid polluting the cache, we consider:

1. The node’s last activity timestamp
2. The graph’s retention period (e.g., 100 days)
3. The cache TTL (e.g., 30 days)

If a node was last active 99 days ago, it expires from the graph in a day, so a 30-day TTL makes no sense, and we skip it. We reserve cache space for active nodes like Account X. This “smart TTL” policy keeps the cache focused on live stories rather than archival ones, so repeat queries for the same part of the graph return faster.

Caching is integrated into the journey, not an afterthought. The engine reuses knowledge from previous queries, so repeated traversals over the same part of the graph keep getting cheaper

The Payoff

The serving layer sits in front of 8 billion nodes and 150 billion edges, serving mixed workloads, all of which need to feel interactive. Single-hop queries return at a P50 of 15–30ms with P99 under 100ms. Even 3-hop traversals, the kind that chain across accounts, profiles, and content, come back at P99 between 100–150ms. Breadth-first execution and parallelism within each level keep these numbers stable even as fan-out grows.

The async-first design is what enables the throughput. Thousands of concurrent requests flow through just 16–24 threads spread across dedicated pools because no thread ever blocks on I/O. When load spikes, our concurrency limiter lets work queue briefly: slowly increasing capacity when things are healthy, backing off aggressively when they’re not

Caching has the most visible impact on day-to-day efficiency. Popular entities like accounts, profiles, and content achieve 70–80% cache hit rates, resulting in roughly 3–4x fewer storage calls on common query paths. Smart TTLs keep the cache focused on active data, avoiding wasted memory on nodes that are near the end of their graph retention window.

These properties, together, make multi-hop graph queries over billions of entities feel, at query time, much closer to in-memory lookups than to remote calls.

What We Learned Along the Way

The biggest surprise wasn’t any single optimization: it was how much async composition changed the economics of our system. We expected it to help latency; we didn’t expect it to slash infrastructure cost. A serving layer that would have needed hundreds of threads per instance runs comfortably on 16–24, because no thread ever blocks on I/O. The tradeoff is debuggability: async stack traces are hard to read, and exceptions can get lost in future chains. We compensated with per-stage metrics, measuring each request at validation, storage, enrichment, and end-to-end, so when something is slow, we know exactly which stage to blame.

Caching took longer to get right than expected. Our first instinct was to cache everything in EVCache and let TTLs handle freshness, but that wastes memory on nodes about to expire from the graph anyway. The breakthrough was matching TTLs to data volatility: stable node properties get long TTLs, while nodes near the end of their retention window aren’t cached at all. The 70–80% hit rate we see today came from being selective, not aggressive.

The filtering hierarchy was born out of frustration. Early on, every new use case meant a code change: one team wanted a 7-day lookback, another 90 days, a third different limits at different depths. Instead of bespoke logic per team, we built a layered override system: application defaults, global overrides, per-depth limits, and per-edge-type limits. It took real effort, but it eliminated an entire class of feature requests and teams now tune their own queries without touching our code.

Closing: Principles for Distributed Systems

The lessons above are specific to the RDG, but the underlying principles apply to any distributed system built around I/O-heavy, fan-out workloads.

• Think in terms of frontiers, not features. Design your APIs so callers describe what frontier to explore, then let the system decide how to walk it efficiently.
• Filter early, not late. Every byte you fetch but don’t need is wasted I/O. Push filters and limits as close to the storage layer as possible: discard irrelevant data at each stage rather than fetching everything and trimming at the end.
• Parallelize deliberately, not by default. Unbounded concurrency feels fast until it overwhelms the systems you depend on. Set explicit limits, monitor them, and adjust dynamically: treat concurrency as a dial, not a switch.
• Treat caching as a first‑class design choice, not an afterthought. Decide what is worth remembering, for how long, and what should be allowed to fade out of memory. Match TTLs to data volatility, and don’t cache what’s about to expire.

—

Thanks for reading Part 3 of the RDG blog series. For us, getting these details right is what turns a constantly changing, billion-edge graph into something that, at query time, feels like a responsive, in-memory data structure.

---

How and Why Netflix Built a Real-Time Distributed Graph: Part 3 — Querying the graph with gRPC… was originally published in Netflix TechBlog on Medium, where people are continuing the conversation by highlighting and responding to this story.