Netflix/Apache Kafka

5 posts

netflix3 min readCurated summary

Building Service Topology at Scale: Architecture, Challenges, and Lessons Learned

The post explains how Netflix built a real-time service topology system capable of processing millions of network-flow records per second at production scale. Its core design combines streaming ingestion, reactive backpressure, physically separate data layers, and a distributed aggregation pipeline that resolves network intermediaries into meaningful service dependencies. The system favors slightly delayed but complete updates over stale batch data or incomplete results caused by dropping records. ## The Need for Real-Time Topology - Traditional topology tools rely on hourly or daily batch processing, making their data outdated during incidents. - Netflix combines: - eBPF network flows - IPC metrics delivered through Server-Sent Events - Distributed tracing data - These sources are stored in separate graph or columnar storage layers and can be queried independently or merged. - The goal is near-real-time freshness, faster incident response, blast-radius analysis, and immediate change validation. ## Backpressure for Reliable Streaming - Processing millions of flow records per second creates a risk that downstream systems will become overwhelmed. - Common alternatives are inadequate: - Unbounded queues eventually exhaust memory. - Dropping records produces incomplete topology. - Batch processing introduces unacceptable delays. - Reactive streams propagate slowdown upstream: - A graph database signals Stage 2. - Stage 2 slows Stage 1. - Stage 1 pauses Kafka consumption. - Kafka retains the data until capacity returns. - This allows the system to degrade gracefully during traffic spikes, garbage-collection pauses, or temporary storage slowdowns. - Updates may be delayed by seconds or minutes, but the data remains substantially more complete than a dropped or hourly-processed stream. ## Physically Separate Topology Layers Netflix keeps each data source in storage optimized for its characteristics: - **Network layer:** eBPF flow logs provide broad coverage but limited application context. - **IPC layer:** Application metrics offer detailed endpoint information but cover only instrumented services. - **Tracing layer:** Parquet-based distributed traces show actual request paths but are sampled. - Separate storage enables each layer to evolve and scale independently. - Queries can run in parallel and merge results while preserving sub-second response times. ## Three-Stage Distributed Aggregation The network layer uses a distributed pipeline to transform individual network hops into logical service dependencies. - Cloud traffic commonly passes through load balancers, NAT gateways, API gateways, and proxies. - Flow logs therefore show relationships such as: - `App A → Load Balancer` - `Load Balancer → App B` - The useful topology must infer the logical dependency: `App A → App B`. ### Stage 1: Initial Flow Aggregation - Consumes flow logs from Kafka across four regions. - Filters invalid records. - Groups data into five-minute windows. - Creates initial aggregators for each window. - Uses consistent hashing to distribute aggregators. - Streams the results to Stage 2 through SSE. ### Stage 2: Intermediary Resolution - Receives the initial aggregators from Stage 1. - Groups flows by intermediary components. - Resolves multi-hop network paths into application-level relationships. - This prevents infrastructure components from dominating the resulting service graph. ## Engineering Trade-offs - Streaming provides much fresher data than batch processing but introduces greater operational and conceptual complexity. - Backpressure is essential for stability at Netflix’s scale, even though reactive pipelines are harder to reason about than synchronous systems. - The architecture prioritizes reliable, complete topology updates over perfectly immediate processing. - Production behavior differed substantially from local testing: consumers lagged, memory was exhausted, traffic became unevenly distributed, and garbage collection consumed significant resources. Netflix’s approach demonstrates that large-scale real-time topology requires streaming ingestion, end-to-end backpressure, specialized storage, and staged aggregation. For similar distributed systems, the practical recommendation is to design explicitly for overload and partial slowdown rather than relying on unbounded buffering, dropped data, or stale batch snapshots.

Read original(opens in new tab)
netflix4 min readCurated summary

Dynamic Repartitioning for Time Series Workloads

Netflix’s TimeSeries Abstraction uses Cassandra to ingest and query petabytes of temporal data with millisecond-scale latency, but growing partitions can cause seconds-long reads, timeouts, and resource exhaustion. Its initial time-based partitioning works well when workload estimates are accurate, yet traffic changes and outlier IDs can make partitions too large or too small. Netflix therefore developed automated time-slice repartitioning and, for isolated hot IDs, asynchronous dynamic partitioning at the individual-ID level. ## Cassandra and the Wide-Partition Problem - Cassandra provides: - High-throughput, low-latency reads and writes - Cost-effective operation at scale - Strong operational familiarity within Netflix - TimeSeries datasets accumulate events over time, creating potentially very wide partitions. - Wide partitions can lead to: - Read latencies increasing from milliseconds to seconds - Request timeouts - Garbage-collection pauses - High CPU utilization and thread queueing - Scaling Cassandra clusters can help, but Netflix sought more targeted solutions. ## Initial Time-Based Partitioning - TimeSeries divides data into discrete time slices to keep partitions manageable. - This structure also makes it efficient to: - Query data by time - Drop old data without creating large tombstone problems - At dataset creation, users provide expected workload characteristics. - Netflix’s provisioning pipeline uses those inputs, along with Monte Carlo simulations, to select infrastructure and partition settings. ## Why Static Provisioning Falls Short - Workloads may be unknown or inaccurately estimated during initial provisioning. - Traffic patterns, client behavior, and product needs can change over time. - A small number of TimeSeries IDs may generate far more events than the rest. - Time slices provide a way to change partitioning for future data, but manually updating thousands of datasets is impractical. ## Repartitioning Entire Time Slices - Cassandra introspection tools, such as `nodetool tablehistograms`, expose partition-size distributions. - Netflix added a background worker that: - Monitors partition histograms for time slices - Publishes observations through a Cassandra virtual table - Detects partitions that are too large or too small - Calculates a new partitioning adjustment factor - Target partition density is typically between 2 MiB and 10 MiB, depending on workload. - The worker updates the strategy for future time slices. For example, it may expand a `time_bucket` interval from 60 seconds to 604,800 seconds when partitions are too small. - This approach reduced read latency and timeouts caused by thread queueing. - Its limitation is that it changes partitioning broadly and is ineffective when only a minority of IDs produce oversized partitions. ## Handling Isolated Problem IDs Netflix considers several responses when only some IDs are problematic: - **Do nothing:** Appropriate when wide partitions do not affect application-level metrics. - **Partial returns:** Abort a request after it exceeds a latency SLO while returning data already collected; useful when latency matters more than completeness. - **Block IDs:** Prevent exceptionally bad test, spam, or otherwise harmful IDs from destabilizing the system. - These options are inadequate when valid, important IDs must return all their data despite generating large partitions. ## Dynamic Partitioning per ID Dynamic partitioning addresses outliers by splitting partitions for individual TimeSeries IDs rather than modifying an entire table. The asynchronous pipeline has three stages: - **Detection:** The read path identifies partitions that exceed a configured size threshold. - **Planning and splitting:** The system asynchronously plans and executes splits into appropriately sized partitions. - **Serving reads:** Once splits are available, read requests are transparently rerouted to them. During each read, the server tracks the bytes retrieved for a partition. If usage exceeds the threshold, it emits a detection event to Kafka containing information such as: - The Cassandra time-slice table - The affected TimeSeries ID - The existing time and event bucket - Whether the partition is immutable - A version identifier ## Practical Recommendation Use whole-time-slice repartitioning when an entire dataset is systematically over- or under-partitioned. For isolated but important high-volume IDs, dynamic per-ID partitioning provides a more precise way to control latency without disrupting the rest of the dataset.

Read original(opens in new tab)
netflix3 min readCurated summary

Democratizing Machine Learning at Netflix: Building the Model Lifecycle Graph

Netflix’s growing use of machine learning across personalization, Studio, payments, advertising, and other domains has created a fragmented ecosystem of tools and metadata. The Metadata Service (MDS) addresses this problem by building a Model Lifecycle Graph that connects models, features, pipelines, experiments, datasets, and ownership information. Its goal is to make ML assets discoverable, understandable, and reusable across organizational boundaries. ## A Fragmented Machine Learning Landscape - Netflix ML has expanded from personalization into areas such as: - Studio production and post-production - Fraud detection and payment optimization - Advertising and real-time targeting - Each domain uses different technologies, metrics, and organizational structures. - Valuable assets often remain isolated in specialized systems. - For example, Studio-generated content embeddings could support: - Contextual ad matching - Episodic merchandising - Recommendations based on tone, topic, or mood - Practitioners struggle to answer basic questions because relevant information is split across: - Model registries - Pipeline orchestrators - Experimentation platforms - Feature stores - Dataset systems - This fragmentation makes discovery, lineage tracking, impact analysis, and ownership difficult. ## The Challenge of Connecting ML Infrastructure - MDS must unify metadata from many independent systems, including: - Pipeline execution and transformation data - Model versions, artifacts, deployments, and staleness - A/B test configurations - Feature definitions and usage - Dataset creation and discovery - User, team, and organization information - These systems use different identifiers, formats, and conceptual models. - The core challenge is transforming heterogeneous metadata into a common entity model and connected graph—not merely creating a consolidated user interface. ## The Model Lifecycle Graph - Netflix’s Metadata Service indexes ML-related assets and materializes relationships between them. - It supports real-time metadata ingestion and cross-domain questions such as: - Which experiments use a particular model? - Which models depend on a feature? - What data sources feed a model? - Who owns each part of the workflow? - The graph is intended to make every ML asset discoverable and reusable regardless of its originating team or business domain. ## Core Concepts and Vocabulary - **Component:** Any uniquely addressable object identified by an AIP URI, such as: - `aip://model/registry/ranking-v5` - `aip://user/identity/alice` - `aip://pipeline/orchestrator/weekly-training` - **Entity:** A component enriched with properties such as name, description, creation date, and ownership. - **Entity type:** A group of entities sharing the same data shape and required properties. - **Domain:** An abstract interface for a category of ML assets, such as Models or Pipelines. - **Provider:** A concrete backend implementation of a domain, such as Netflix’s internal model registry. - Separating domains from providers allows multiple systems to implement the same interface without changing how consumers interact with MDS. - URI-based addressing gives services a consistent way to reference assets and resolve them to connected metadata. ## From Events to a Queryable Graph - MDS receives metadata events through Kafka and AWS SNS/SQS. - Source systems emit lightweight events containing an event type and resource identifier. - For example, a model registry might emit a `model_instance_created` event with the new instance’s ID. - This keeps event producers simple while allowing MDS to enrich events, construct entities, and infer relationships such as connections between models and A/B tests. The Model Lifecycle Graph provides Netflix with a common layer for connecting previously isolated ML systems. By standardizing identifiers, entities, domains, and providers, MDS can support cross-domain discovery, lineage, impact analysis, and collaboration at scale.

Read original(opens in new tab)
netflix3 min readCurated summary

Powering Multimodal Intelligence for Video Search

Video search is difficult because it must combine many kinds of information—characters, scenes, dialogue, labels, and embeddings—across enormous volumes of footage. The post argues that solving this problem requires a distributed pipeline that separates reliable ingestion, computationally intensive data fusion, and low-latency search indexing. Temporal bucketing, hybrid ranking, and deduplication turn billions of model outputs into searchable moments for editors. ## Why Video Search Is Complex - Video contains multiple overlapping modalities, each analyzed by specialized models. - Models produce different outputs, including: - Text labels such as characters or objects - Scene classifications - High-dimensional embedding vectors - Time ranges with varying boundaries - Overlapping model timelines must be synchronized into a chronological representation. - A 2,000-hour archive may contain more than 216 million frames, expanding to billions of records after multimodal processing. - Search must avoid returning thousands of redundant clips from continuous shots. - Ranking therefore combines: - Symbolic text matching for precision and interpretability - Semantic vector similarity for contextual relevance - Clustering and deduplication to identify the best moments - Sub-second response times are essential because delays interrupt editors’ creative workflows. ## Three-Stage Ingestion and Fusion Pipeline ### Transactional Persistence - Raw model annotations are ingested through highly available pipelines. - Apache Cassandra stores the annotations with an emphasis on: - Data integrity - Distributed availability - High write throughput - An annotation can include a type, nanosecond time range, embedding vector, label, and confidence score. ### Offline Data Fusion - After persistence, Apache Kafka publishes an event that starts asynchronous processing. - The offline pipeline performs expensive temporal intersections without slowing ingestion or search. - Model outputs are normalized into fixed one-second time buckets. - The fusion process: - Maps continuous detections into discrete intervals - Intersects annotations sharing a bucket - Combines them into unified records - Writes the enriched records back to Cassandra - For example, a “Joey” character detection from seconds 2–8 can be combined with a “kitchen” scene detection from seconds 4–9 to create a fused record for the 4–5 second interval. - Each fused record retains links to the original annotations and source asset. ### Real-Time Search Indexing - Enriched buckets are later sent from Cassandra to Elasticsearch. - Upserts use a composite key consisting of the asset ID and time bucket. - If a bucket already exists, it is updated rather than duplicated. - This creates one consistent record for each second of footage while allowing new model results to be incorporated. The overall recommendation is to treat multimodal video search as a distributed data-fusion problem rather than a single-model retrieval task. Decoupling ingestion, offline processing, and indexing allows the system to handle massive archives while preserving reliable data capture and fast, context-rich search.

Read original(opens in new tab)
netflixOriginal article

Building a Resilient Data Platform with Write-Ahead Log at Netflix | by Netflix Technology Blog | Netflix TechBlog (opens in new tab)

Netflix has developed a distributed Write-Ahead Log (WAL) abstraction to address critical data challenges such as accidental corruption, system entropy, and the complexities of cross-region replication. By decoupling data mutation from immediate persistence and providing a unified API, this system ensures strong durability and eventual consistency across diverse storage engines. The WAL acts as a resilient buffer that powers high-leverage features like secondary indexing and delayed retry queues while maintaining the massive scale required for global operations. ### The Role of the WAL Abstraction * The system serves as a centralized mechanism to capture data changes and reliably deliver them to downstream consumers, mitigating the risk of data loss during administrative errors or database corruption. * It provides a simplified `WriteToLog` gRPC endpoint that abstracts underlying infrastructure, allowing developers to focus on data logic rather than the specifics of the storage layer. * By acting as a durable intermediary, it prevents permanent data loss during incidents where primary datastores fail or require schema changes that might otherwise lead to corruption. ### Flexible Personas and Namespaces * The architecture utilizes "namespaces" to define logical separation, allowing different services to configure specific storage backends like Kafka or SQS based on their needs. * The "Delayed Queues" persona leverages SQS to provide a scalable way to retry failed messages in real-time pipelines without sacrificing overall system throughput. * The system can be configured for "Cross-Region Replication," enabling high availability and disaster recovery for storage engines that do not natively support multi-region data transfer. ### Solving System Entropy and Consistency * The WAL addresses the "dual-write" problem, where updates to primary stores (such as Cassandra) and search indices (such as Elasticsearch) can diverge over time, leading to data inconsistency. * It facilitates reliable secondary indexing for NoSQL databases by managing updates to multiple partitions as a coordinated sequence of events. * The platform mitigates operational risks, such as Out-of-Memory (OOM) errors on Key-Value nodes caused by bulk deletes, by staging and throttling mutations through the log. Organizations operating at scale should adopt a WAL-centric architecture to simplify the management of heterogeneous data stores and enhance system resilience. By centralizing the mutation log, teams can implement complex features like Change Data Capture (CDC) and cross-region failover through a single, consistent interface rather than building bespoke solutions for every service.