By: Rajiv Shringi, Oleksii Tkachuk, Kartik Sathyanarayanan
In our earlier weblog put up, we launched Netflix’s TimeSeries Abstraction, a distributed service designed to retailer and question massive volumes of temporal occasion knowledge with low millisecond latencies. Immediately, we’re excited to current the Distributed Counter Abstraction. This counting service, constructed on prime of the TimeSeries Abstraction, allows distributed counting at scale whereas sustaining comparable low latency efficiency. As with all our abstractions, we use our Information Gateway Management Airplane to shard, configure, and deploy this service globally.
Distributed counting is a difficult drawback in pc science. On this weblog put up, we’ll discover the various counting necessities at Netflix, the challenges of reaching correct counts in close to real-time, and the rationale behind our chosen strategy, together with the mandatory trade-offs.
Notice: In the case of distributed counters, phrases similar to ‘correct’ or ‘exact’ ought to be taken with a grain of salt. On this context, they discuss with a depend very near correct, offered with minimal delays.
At Netflix, our counting use circumstances embrace monitoring tens of millions of consumer interactions, monitoring how typically particular options or experiences are proven to customers, and counting a number of aspects of information throughout A/B take a look at experiments, amongst others.
At Netflix, these use circumstances will be labeled into two broad classes:
- Greatest-Effort: For this class, the depend doesn’t should be very correct or sturdy. Nonetheless, this class requires near-immediate entry to the present depend at low latencies, all whereas conserving infrastructure prices to a minimal.
- Finally Constant: This class wants correct and sturdy counts, and is prepared to tolerate a slight delay in accuracy and a barely greater infrastructure value as a trade-off.
Each classes share widespread necessities, similar to excessive throughput and excessive availability. The desk under offers an in depth overview of the various necessities throughout these two classes.
To fulfill the outlined necessities, the Counter Abstraction was designed to be extremely configurable. It permits customers to decide on between completely different counting modes, similar to Greatest-Effort or Finally Constant, whereas contemplating the documented trade-offs of every possibility. After deciding on a mode, customers can work together with APIs without having to fret concerning the underlying storage mechanisms and counting strategies.
Let’s take a more in-depth take a look at the construction and performance of the API.
Counters are organized into separate namespaces that customers arrange for every of their particular use circumstances. Every namespace will be configured with completely different parameters, similar to Kind of Counter, Time-To-Reside (TTL), and Counter Cardinality, utilizing the service’s Management Airplane.
The Counter Abstraction API resembles Java’s AtomicInteger interface:
AddCount/AddAndGetCount: Adjusts the depend for the desired counter by the given delta worth inside a dataset. The delta worth will be optimistic or adverse. The AddAndGetCount counterpart additionally returns the depend after performing the add operation.
{
"namespace": "my_dataset",
"counter_name": "counter123",
"delta": 2,
"idempotency_token": {
"token": "some_event_id",
"generation_time": "2024-10-05T14:48:00Z"
}
}The idempotency token can be utilized for counter sorts that assist them. Purchasers can use this token to soundly retry or hedge their requests. Failures in a distributed system are a given, and being able to soundly retry requests enhances the reliability of the service.
GetCount: Retrieves the depend worth of the desired counter inside a dataset.
{
"namespace": "my_dataset",
"counter_name": "counter123"
}ClearCount: Successfully resets the depend to 0 for the desired counter inside a dataset.
{
"namespace": "my_dataset",
"counter_name": "counter456",
"idempotency_token": {...}
}Now, let’s take a look at the several types of counters supported inside the Abstraction.
The service primarily helps two kinds of counters: Greatest-Effort and Finally Constant, together with a 3rd experimental kind: Correct. Within the following sections, we’ll describe the completely different approaches for a lot of these counters and the trade-offs related to every.
Such a counter is powered by EVCache, Netflix’s distributed caching resolution constructed on the broadly widespread Memcached. It’s appropriate to be used circumstances like A/B experiments, the place many concurrent experiments are run for comparatively brief durations and an approximate depend is enough. Setting apart the complexities of provisioning, useful resource allocation, and management airplane administration, the core of this resolution is remarkably simple:
// counter cache key
counterCacheKey = <namespace>:<counter_name>// add operation
return delta > 0
? cache.incr(counterCacheKey, delta, TTL)
: cache.decr(counterCacheKey, Math.abs(delta), TTL);
// get operation
cache.get(counterCacheKey);
// clear counts from all replicas
cache.delete(counterCacheKey, ReplicaPolicy.ALL);
EVCache delivers extraordinarily excessive throughput at low millisecond latency or higher inside a single area, enabling a multi-tenant setup inside a shared cluster, saving infrastructure prices. Nonetheless, there are some trade-offs: it lacks cross-region replication for the increment operation and doesn’t present consistency ensures, which can be crucial for an correct depend. Moreover, idempotency isn’t natively supported, making it unsafe to retry or hedge requests.
Whereas some customers might settle for the restrictions of a Greatest-Effort counter, others go for exact counts, sturdiness and world availability. Within the following sections, we’ll discover varied methods for reaching sturdy and correct counts. Our goal is to focus on the challenges inherent in world distributed counting and clarify the reasoning behind our chosen strategy.
Strategy 1: Storing a Single Row per Counter
Let’s begin easy through the use of a single row per counter key inside a desk in a globally replicated datastore.
Let’s look at a number of the drawbacks of this strategy:
- Lack of Idempotency: There isn’t a idempotency key baked into the storage data-model stopping customers from safely retrying requests. Implementing idempotency would doubtless require utilizing an exterior system for such keys, which might additional degrade efficiency or trigger race circumstances.
- Heavy Competition: To replace counts reliably, each author should carry out a Examine-And-Swap operation for a given counter utilizing locks or transactions. Relying on the throughput and concurrency of operations, this may result in important competition, closely impacting efficiency.
Secondary Keys: One strategy to scale back competition on this strategy can be to make use of a secondary key, similar to a bucket_id, which permits for distributing writes by splitting a given counter into buckets, whereas enabling reads to combination throughout buckets. The problem lies in figuring out the suitable variety of buckets. A static quantity should still result in competition with sizzling keys, whereas dynamically assigning the variety of buckets per counter throughout tens of millions of counters presents a extra advanced drawback.
Let’s see if we are able to iterate on our resolution to beat these drawbacks.
Strategy 2: Per Occasion Aggregation
To deal with problems with sizzling keys and competition from writing to the identical row in real-time, we may implement a method the place every occasion aggregates the counts in reminiscence after which flushes them to disk at common intervals. Introducing enough jitter to the flush course of can additional scale back competition.
Nonetheless, this resolution presents a brand new set of points:
- Vulnerability to Information Loss: The answer is weak to knowledge loss for all in-memory knowledge throughout occasion failures, restarts, or deployments.
- Incapacity to Reliably Reset Counts: On account of counting requests being distributed throughout a number of machines, it’s difficult to ascertain consensus on the precise cut-off date when a counter reset occurred.
- Lack of Idempotency: Just like the earlier strategy, this methodology doesn’t natively assure idempotency. One strategy to obtain idempotency is by constantly routing the identical set of counters to the identical occasion. Nonetheless, this strategy might introduce further complexities, similar to chief election, and potential challenges with availability and latency within the write path.
That mentioned, this strategy should still be appropriate in eventualities the place these trade-offs are acceptable. Nonetheless, let’s see if we are able to tackle a few of these points with a unique event-based strategy.
Strategy 3: Utilizing Sturdy Queues
On this strategy, we log counter occasions right into a sturdy queuing system like Apache Kafka to forestall any potential knowledge loss. By creating a number of matter partitions and hashing the counter key to a particular partition, we be sure that the identical set of counters are processed by the identical set of customers. This setup simplifies facilitating idempotency checks and resetting counts. Moreover, by leveraging further stream processing frameworks similar to Kafka Streams or Apache Flink, we are able to implement windowed aggregations.
Nonetheless, this strategy comes with some challenges:
- Potential Delays: Having the identical client course of all of the counts from a given partition can result in backups and delays, leading to stale counts.
- Rebalancing Partitions: This strategy requires auto-scaling and rebalancing of matter partitions because the cardinality of counters and throughput will increase.
Moreover, all approaches that pre-aggregate counts make it difficult to assist two of our necessities for correct counters:
- Auditing of Counts: Auditing entails extracting knowledge to an offline system for evaluation to make sure that increments had been utilized accurately to achieve the ultimate worth. This course of can be used to trace the provenance of increments. Nonetheless, auditing turns into infeasible when counts are aggregated with out storing the person increments.
- Potential Recounting: Just like auditing, if changes to increments are crucial and recounting of occasions inside a time window is required, pre-aggregating counts makes this infeasible.
Barring these few necessities, this strategy can nonetheless be efficient if we decide the precise strategy to scale our queue partitions and customers whereas sustaining idempotency. Nonetheless, let’s discover how we are able to alter this strategy to satisfy the auditing and recounting necessities.
Strategy 4: Occasion Log of Particular person Increments
On this strategy, we log every particular person counter increment together with its event_time and event_id. The event_id can embrace the supply info of the place the increment originated. The mixture of event_time and event_id can even function the idempotency key for the write.
Nonetheless, in its easiest type, this strategy has a number of drawbacks:
- Learn Latency: Every learn request requires scanning all increments for a given counter doubtlessly degrading efficiency.
- Duplicate Work: A number of threads would possibly duplicate the trouble of aggregating the identical set of counters throughout learn operations, resulting in wasted effort and subpar useful resource utilization.
- Extensive Partitions: If utilizing a datastore like Apache Cassandra, storing many increments for a similar counter may result in a large partition, affecting learn efficiency.
- Giant Information Footprint: Storing every increment individually may additionally lead to a considerable knowledge footprint over time. With out an environment friendly knowledge retention technique, this strategy might battle to scale successfully.
The mixed impression of those points can result in elevated infrastructure prices that could be tough to justify. Nonetheless, adopting an event-driven strategy appears to be a big step ahead in addressing a number of the challenges we’ve encountered and assembly our necessities.
How can we enhance this resolution additional?
We use a mix of the earlier approaches, the place we log every counting exercise as an occasion, and repeatedly combination these occasions within the background utilizing queues and a sliding time window. Moreover, we make use of a bucketing technique to forestall large partitions. Within the following sections, we’ll discover how this strategy addresses the beforehand talked about drawbacks and meets all our necessities.
Notice: From right here on, we’ll use the phrases “rollup” and “combination” interchangeably. They basically imply the identical factor, i.e., accumulating particular person counter increments/decrements and arriving on the last worth.
TimeSeries Occasion Retailer:
We selected the TimeSeries Information Abstraction as our occasion retailer, the place counter mutations are ingested as occasion information. A number of the advantages of storing occasions in TimeSeries embrace:
Excessive-Efficiency: The TimeSeries abstraction already addresses a lot of our necessities, together with excessive availability and throughput, dependable and quick efficiency, and extra.
Lowering Code Complexity: We scale back lots of code complexity in Counter Abstraction by delegating a serious portion of the performance to an present service.
TimeSeries Abstraction makes use of Cassandra because the underlying occasion retailer, however it may be configured to work with any persistent retailer. Here’s what it seems to be like:
Dealing with Extensive Partitions: The time_bucket and event_bucket columns play a vital function in breaking apart a large partition, stopping high-throughput counter occasions from overwhelming a given partition. For extra info relating to this, discuss with our earlier weblog.
No Over-Counting: The event_time, event_id and event_item_key columns type the idempotency key for the occasions for a given counter, enabling purchasers to retry safely with out the danger of over-counting.
Occasion Ordering: TimeSeries orders all occasions in descending order of time permitting us to leverage this property for occasions like depend resets.
Occasion Retention: The TimeSeries Abstraction consists of retention insurance policies to make sure that occasions aren’t saved indefinitely, saving disk house and decreasing infrastructure prices. As soon as occasions have been aggregated and moved to a more cost effective retailer for audits, there’s no must retain them within the main storage.
Now, let’s see how these occasions are aggregated for a given counter.
Aggregating Rely Occasions:
As talked about earlier, accumulating all particular person increments for each learn request can be cost-prohibitive by way of learn efficiency. Due to this fact, a background aggregation course of is critical to repeatedly converge counts and guarantee optimum learn efficiency.
However how can we safely combination depend occasions amidst ongoing write operations?
That is the place the idea of Finally Constant counts turns into essential. By deliberately lagging behind the present time by a secure margin, we be sure that aggregation all the time happens inside an immutable window.
Lets see what that appears like:
Let’s break this down:
- lastRollupTs: This represents the latest time when the counter worth was final aggregated. For a counter being operated for the primary time, this timestamp defaults to an inexpensive time prior to now.
- Immutable Window and Lag: Aggregation can solely happen safely inside an immutable window that’s now not receiving counter occasions. The “acceptLimit” parameter of the TimeSeries Abstraction performs a vital function right here, because it rejects incoming occasions with timestamps past this restrict. Throughout aggregations, this window is pushed barely additional again to account for clock skews.
This does imply that the counter worth will lag behind its most up-to-date replace by some margin (sometimes within the order of seconds). This strategy does go away the door open for missed occasions as a result of cross-region replication points. See “Future Work” part on the finish.
- Aggregation Course of: The rollup course of aggregates all occasions within the aggregation window for the reason that final rollup to reach on the new worth.
Rollup Retailer:
We save the outcomes of this aggregation in a persistent retailer. The following aggregation will merely proceed from this checkpoint.
We create one such Rollup desk per dataset and use Cassandra as our persistent retailer. Nonetheless, as you’ll quickly see within the Management Airplane part, the Counter service will be configured to work with any persistent retailer.
LastWriteTs: Each time a given counter receives a write, we additionally log a last-write-timestamp as a columnar replace on this desk. That is performed utilizing Cassandra’s USING TIMESTAMP function to predictably apply the Final-Write-Win (LWW) semantics. This timestamp is similar because the event_time for the occasion. Within the subsequent sections, we’ll see how this timestamp is used to maintain some counters in energetic rollup circulation till they’ve caught as much as their newest worth.
Rollup Cache
To optimize learn efficiency, these values are cached in EVCache for every counter. We mix the lastRollupCount and lastRollupTs right into a single cached worth per counter to forestall potential mismatches between the depend and its corresponding checkpoint timestamp.
However, how do we all know which counters to set off rollups for? Let’s discover our Write and Learn path to know this higher.
Add/Clear Rely:
An add or clear depend request writes durably to the TimeSeries Abstraction and updates the last-write-timestamp within the Rollup retailer. If the sturdiness acknowledgement fails, purchasers can retry their requests with the identical idempotency token with out the danger of overcounting. Upon sturdiness, we ship a fire-and-forget request to set off the rollup for the request counter.
GetCount:
We return the final rolled-up depend as a fast point-read operation, accepting the trade-off of probably delivering a barely stale depend. We additionally set off a rollup through the learn operation to advance the last-rollup-timestamp, enhancing the efficiency of subsequent aggregations. This course of additionally self-remediates a stale depend if any earlier rollups had failed.
With this strategy, the counts regularly converge to their newest worth. Now, let’s see how we scale this strategy to tens of millions of counters and 1000’s of concurrent operations utilizing our Rollup Pipeline.
Rollup Pipeline:
Every Counter-Rollup server operates a rollup pipeline to effectively combination counts throughout tens of millions of counters. That is the place a lot of the complexity in Counter Abstraction is available in. Within the following sections, we’ll share key particulars on how environment friendly aggregations are achieved.
Gentle-Weight Roll-Up Occasion: As seen in our Write and Learn paths above, each operation on a counter sends a lightweight occasion to the Rollup server:
rollupEvent: {
"namespace": "my_dataset",
"counter": "counter123"
}Notice that this occasion doesn’t embrace the increment. That is solely a sign to the Rollup server that this counter has been accessed and now must be aggregated. Understanding precisely which particular counters should be aggregated prevents scanning your entire occasion dataset for the aim of aggregations.
In-Reminiscence Rollup Queues: A given Rollup server occasion runs a set of in-memory queues to obtain rollup occasions and parallelize aggregations. Within the first model of this service, we settled on utilizing in-memory queues to cut back provisioning complexity, save on infrastructure prices, and make rebalancing the variety of queues pretty simple. Nonetheless, this comes with the trade-off of probably lacking rollup occasions in case of an occasion crash. For extra particulars, see the “Stale Counts” part in “Future Work.”
Decrease Duplicate Effort: We use a quick non-cryptographic hash like XXHash to make sure that the identical set of counters find yourself on the identical queue. Additional, we attempt to reduce the quantity of duplicate aggregation work by having a separate rollup stack that chooses to run fewer beefier situations.
Availability and Race Situations: Having a single Rollup server occasion can reduce duplicate aggregation work however might create availability challenges for triggering rollups. If we select to horizontally scale the Rollup servers, we permit threads to overwrite rollup values whereas avoiding any type of distributed locking mechanisms to keep up excessive availability and efficiency. This strategy stays secure as a result of aggregation happens inside an immutable window. Though the idea of now() might differ between threads, inflicting rollup values to generally fluctuate, the counts will ultimately converge to an correct worth inside every immutable aggregation window.
Rebalancing Queues: If we have to scale the variety of queues, a easy Management Airplane configuration replace adopted by a re-deploy is sufficient to rebalance the variety of queues.
"eventual_counter_config": {
"queue_config": {
"num_queues" : 8, // change to 16 and re-deploy
...Dealing with Deployments: Throughout deployments, these queues shut down gracefully, draining all present occasions first, whereas the brand new Rollup server occasion begins up with doubtlessly new queue configurations. There could also be a short interval when each the previous and new Rollup servers are energetic, however as talked about earlier than, this race situation is managed since aggregations happen inside immutable home windows.
Decrease Rollup Effort: Receiving a number of occasions for a similar counter doesn’t imply rolling it up a number of occasions. We drain these rollup occasions right into a Set, guaranteeing a given counter is rolled up solely as soon as throughout a rollup window.
Environment friendly Aggregation: Every rollup client processes a batch of counters concurrently. Inside every batch, it queries the underlying TimeSeries abstraction in parallel to combination occasions inside specified time boundaries. The TimeSeries abstraction optimizes these vary scans to realize low millisecond latencies.
Dynamic Batching: The Rollup server dynamically adjusts the variety of time partitions that should be scanned based mostly on cardinality of counters with the intention to stop overwhelming the underlying retailer with many parallel learn requests.
Adaptive Again-Stress: Every client waits for one batch to finish earlier than issuing the rollups for the subsequent batch. It adjusts the wait time between batches based mostly on the efficiency of the earlier batch. This strategy offers back-pressure throughout rollups to forestall overwhelming the underlying TimeSeries retailer.
Dealing with Convergence:
As a way to stop low-cardinality counters from lagging behind an excessive amount of and subsequently scanning too many time partitions, they’re saved in fixed rollup circulation. For high-cardinality counters, repeatedly circulating them would eat extreme reminiscence in our Rollup queues. That is the place the last-write-timestamp talked about beforehand performs a vital function. The Rollup server inspects this timestamp to find out if a given counter must be re-queued, guaranteeing that we proceed aggregating till it has absolutely caught up with the writes.
Now, let’s see how we leverage this counter kind to offer an up-to-date present depend in near-realtime.
We’re experimenting with a barely modified model of the Finally Constant counter. Once more, take the time period ‘Correct’ with a grain of salt. The important thing distinction between such a counter and its counterpart is that the delta, representing the counts for the reason that last-rolled-up timestamp, is computed in real-time.
Aggregating this delta in real-time can impression the efficiency of this operation, relying on the variety of occasions and partitions that should be scanned to retrieve this delta. The identical precept of rolling up in batches applies right here to forestall scanning too many partitions in parallel.
Conversely, if the counters on this dataset are accessed continuously, the time hole for the delta stays slender, making this strategy of fetching present counts fairly efficient.
Now, let’s see how all this complexity is managed by having a unified Management Airplane configuration.
The Information Gateway Platform Management Airplane manages management settings for all abstractions and namespaces, together with the Counter Abstraction. Under, is an instance of a management airplane configuration for a namespace that helps ultimately constant counters with low cardinality:
"persistence_configuration": [
{
"id": "CACHE", // Counter cache config
"scope": "dal=counter",
"physical_storage": {
"type": "EVCACHE", // type of cache storage
"cluster": "evcache_dgw_counter_tier1" // Shared EVCache cluster
}
},
{
"id": "COUNTER_ROLLUP",
"scope": "dal=counter", // Counter abstraction config
"physical_storage": {
"type": "CASSANDRA", // type of Rollup store
"cluster": "cass_dgw_counter_uc1", // physical cluster name
"dataset": "my_dataset_1" // namespace/dataset
},
"counter_cardinality": "LOW", // supported counter cardinality
"config": {
"counter_type": "EVENTUAL", // Type of counter
"eventual_counter_config": { // eventual counter type
"internal_config": {
"queue_config": { // adjust w.r.t cardinality
"num_queues" : 8, // Rollup queues per instance
"coalesce_ms": 10000, // coalesce duration for rollups
"capacity_bytes": 16777216 // allocated memory per queue
},
"rollup_batch_count": 32 // parallelization factor
}
}
}
},
{
"id": "EVENT_STORAGE",
"scope": "dal=ts", // TimeSeries Event store
"physical_storage": {
"type": "CASSANDRA", // persistent store type
"cluster": "cass_dgw_counter_uc1", // physical cluster name
"dataset": "my_dataset_1", // keyspace name
},
"config": {
"time_partition": { // time-partitioning for events
"buckets_per_id": 4, // event buckets within
"seconds_per_bucket": "600", // smaller width for LOW card
"seconds_per_slice": "86400", // width of a time slice table
},
"accept_limit": "5s", // boundary for immutability
},
"lifecycleConfigs": {
"lifecycleConfig": [
{
"type": "retention", // Event retention
"config": {
"close_after": "518400s",
"delete_after": "604800s" // 7 day count event retention
}
}
]
}
}
]Utilizing such a management airplane configuration, we compose a number of abstraction layers utilizing containers deployed on the identical host, with every container fetching configuration particular to its scope.
As with the TimeSeries abstraction, our automation makes use of a bunch of consumer inputs relating to their workload and cardinalities to reach on the proper set of infrastructure and associated management airplane configuration. You may study extra about this course of in a chat given by one among our gorgeous colleagues, Joey Lynch : How Netflix optimally provisions infrastructure within the cloud.
On the time of penning this weblog, this service was processing near 75K depend requests/second globally throughout the completely different API endpoints and datasets:
whereas offering single-digit millisecond latencies for all its endpoints:
Whereas our system is strong, we nonetheless have work to do in making it extra dependable and enhancing its options. A few of that work consists of:
- Regional Rollups: Cross-region replication points may end up in missed occasions from different areas. An alternate technique entails establishing a rollup desk for every area, after which tallying them in a worldwide rollup desk. A key problem on this design can be successfully speaking the clearing of the counter throughout areas.
- Error Detection and Stale Counts: Excessively stale counts can happen if rollup occasions are misplaced or if a rollups fails and isn’t retried. This isn’t a problem for continuously accessed counters, as they continue to be in rollup circulation. This difficulty is extra pronounced for counters that aren’t accessed continuously. Usually, the preliminary learn for such a counter will set off a rollup, self-remediating the problem. Nonetheless, to be used circumstances that can’t settle for doubtlessly stale preliminary reads, we plan to implement improved error detection, rollup handoffs, and sturdy queues for resilient retries.
Distributed counting stays a difficult drawback in pc science. On this weblog, we explored a number of approaches to implement and deploy a Counting service at scale. Whereas there could also be different strategies for distributed counting, our objective has been to ship blazing quick efficiency at low infrastructure prices whereas sustaining excessive availability and offering idempotency ensures. Alongside the way in which, we make varied trade-offs to satisfy the various counting necessities at Netflix. We hope you discovered this weblog put up insightful.
Keep tuned for Half 3 of Composite Abstractions at Netflix, the place we’ll introduce our Graph Abstraction, a brand new service being constructed on prime of the Key-Worth Abstraction and the TimeSeries Abstraction to deal with high-throughput, low-latency graphs.
Particular due to our gorgeous colleagues who contributed to the Counter Abstraction’s success: Joey Lynch, Vinay Chella, Kaidan Fullerton, Tom DeVoe, Mengqing Wang
