Distributed Recoverable Sketches

Count-Min Sketch (CMS) [8], is a probabilistic data structure for summarizing data streams, which can be used to identify heavy hitters (frequent items). Using several hash functions and corresponding arrays of counters, it responds to a point query with a probably approximately correct answer by counting first and computing the minimum next. This ensures that the overestimation caused by hash collisions is minimal.

Given user-specified parameters $(ϵ,\delta )$, a CMS is represented by a matrix of initially zeroed counters, ${\mathrm{𝐶𝑜𝑢𝑛𝑡}}_{d\times w}$, where $d=\lceil \ln \frac{1}{\delta }\rceil$ and $w=\lceil \frac{e}{ϵ}\rceil$. Additionally, $d$ hash functions $h_1\mathrm{\dots }{h}_d:\{1\mathrm{\dots }n\}\to \{1\mathrm{\dots }w\}$ are chosen uniformly at random from a pairwise-independent family, mapping each item to one cell, for each row of $\mathrm{𝐶𝑜𝑢𝑛𝑡}$.

When an update $(x,c_t)$ arrives, meaning that item $x$ is updated by a quantity of $c_t$, then $c_t$ is added to the corresponding counters, determined by $h_i(x)$ for each row $i$. Formally, $\forall 1\le i\le d$ set $\mathrm{𝐶𝑜𝑢𝑛𝑡}[i,h_i(x)]\leftarrow \mathrm{𝐶𝑜𝑢𝑛𝑡}[i,h_i(x)]+c_t$.

The answer to a point query $Q(x)$ is given by $\widehat{c_x}={\min }_i\mathrm{𝐶𝑜𝑢𝑛𝑡}[i,h_i(x)]$. The estimate $\widehat{c_x}$ has the following guarantees: ($i$) $c_x\le \widehat{c_x}$, where $c_x$ is the true frequency of $x$ within the stream, and ($ii$) with probability at least $(1-\delta )$, $\widehat{c_x}\le c_x+ϵN$, where $N={\sum }_{x=1}^n|c_x|$ is the total number of items in the stream. Alternatively, the probability to exceed the additive error $ϵN$ is bounded by $\delta$, i.e., .

Chen, Lee, Gibson, Katz and Petterson described in 1994 the seven disk array architectures called Redundant Arrays of Inexpensive Disks, aka RAID levels 0-6 [5]. These schemes were developed based on the two architectural techniques used in disk arrays: striping across multiple disks to improve performance, and redundancy to improve reliability. Data striping results in uniform load balancing across all of the disks, while redundancy in the form of error-correcting codes (like parity, Hamming or Reed-Solomon codes) tolerates disk failure. Parity information can be used to recover from any single disk failure. RAID 4 concentrates the parity on dedicated redundant disk, while RAID 5 distributes the parity uniformly across the disk array. Thinking of a redundant disk as holding the sum of all the contents in data disks, when a data disk fails, all the data on non-failed disks can be subtracted from that sum, and the remaining information must be the missing one. RAID 6, also called $P+Q$ redundancy, uses Reed-Solomon codes to tolerate up to two disk failures.

Our work is inspired by RAID, but optimized to the realm of sketches, which in networking applications, must cope with extremely fast update rates across the network. In RAID, the parity can be either reconstructed from scratch or updated incrementally by applying the differences in parity between the new and old data. This general technique can be applied to sketches that process data streams and send periodic updates in cycles. It can be further optimized with compact encoding to reduce the communication cost when only few changes occurred. Yet, two data sketches are required to compute the differences in parity: the original sketch, whose content is updated during the normal execution, and the latest backup from the prior update cycle, aka the new and old data.

As our work is sketch-aware, we use simplified design focusing on integers instead of parity bits. With linear sketches we are able to update the redundant nodes in incremental manner, producing the sums as if were calculated from scratch. To this end, next to its original sketch, each data node holds only the ongoing changes of a cycle.

We consider the general message-passing model with $send$ and $receive$ operations¹¹1Alternatively, one can think of an RDMA-like $write$ and $read$ model., and assume that messages are sent over reliable communication links, similar to TCP/IP or QUIC, that provide an end-to-end service to applications running on nodes. Data nodes could update the redundant nodes on each item arrival by simply forwarding, as they process their data streams. An IP packet is the smallest message entity exchanged via the Internet Protocol across IP network. As each message consists of a header and data (payload), it is reasonable to send updates whenever there are enough changes to fill an IP packet. This would increase network throughput by reducing the communication overhead, associated with transmitting a large number of small frames. Batching and buffering are commonly used techniques, where the events are buffered first and sent as a batch operation next. With batch publishing strategy, RabbitMQ publisher [12] batches the messages before sending them to consumers, reducing the overhead of processing each message individually.

To maintain the redundant information efficiently, we consider periodic updates of incremental changes and examine several data structures to economically represent and encode a batch. Batch may contain only a small portion of overall flows, making it feasible to maintain exact counter-per-flow (e.g., using a compact hash table [1]). We begin with traditional buffering of items as they arrive, and then switch to flows, grouping the items by their identifiers while maintaining their accumulated frequencies using key-value pairs.

Dictionaries are commonly used to implement sets, using an associative array of slots or buckets to store key-value pairs, with the constraint of unique keys. The most efficient dictionaries are based on hashing techniques [15], but hash collisions may occur when more than one key is mapped to the same slot of a hash table. Different implementations exist aiming to provide good performance at very high table load factors. A load factor of a hash table is defined as the ratio between the number of elements occupied in a hash table and its capacity, signifying how full a hash table is. A hash table has to rehash the values once its load factor reaches the load threshold, i.e., the maximum load factor, which is an implementation-specific parameter. To avoid collisions, some extra space is required in the hash table, by having the capacity greater than the number of elements at all times.

We consider an abstract compact hash table having $\alpha$ load threshold. High-performance hash tables often rely on bucketized cuckoo hash table, due to featuring an excellent read performance by guaranteeing that the value associated with some key can be found in less than three memory accesses [16]. While the original Cuckoo hashing [15] requires the average load factor to be kept less than $0.5$, in Blocked cuckoo hashing [9], using four keys per bucket increases the threshold to $0.98$.

Using this strategy, in addition to $k$ data nodes, $r$ dedicated redundant nodes are allocated as well, resulting in overall $(k+r)$ nodes in the system. For $f=1$, similar to RAID 4, a single dedicated redundant node is sufficient to hold a sum of all $k$ data nodes’ sketches. Denote $R={\sum }_{i=1}^k{D}_i$ the sum sketch held by the dedicated redundant node. To recover the sketch of a failed data node $i$, we simply compute $D_i=R-{\sum }_{j\ne i}{D}_j$, where the plus and minus operations are performed as matrix operations.

For any $f$, our goal is to be able to recover the sketches held by the $k$ data nodes, despite the failure of any $f$ nodes, or less. Once the failed data nodes are recovered, the failed redundancy can be reconstructed. Turning to linear algebra, this can be realized by generating a system of $(k+r)$ linear equations with $k$ variables. For a unique solution to exist, a set of vectors that span ${ℝ}^k$ is required, implying that any $k$ linearly independent vectors must be available at all times.

With this strategy, the redundancy is distributed across the $k$ data nodes, each of which can additionally serve as a redundant node. This time, if a node goes down, both its original sketch (aka data) and the redundant information, it held for others, fail. As before, we recover the data first and reconstruct the redundancy next.

Since all the nodes are data nodes, we need more redundant sketches than with dedicated redundancy, i.e., $space\ge (f+1)$. For the base case, $f=1$, in order to distinguish between the roles of a node $i$, denote $D_i$ the original (data) sketch and $R_i$ the redundant sketch held by $i$. $R_i$ may cover no more than the rest of $(k-1)$ data nodes. To be able to recover from $i$’s failure as well, some other node $j$ must cover $D_i$ in its redundant sketch $R_j$. Hence, using this strategy, even for a single failure tolerance, at least two redundant sketches are required.

When $B=1$, no delay is activated, and we update the redundant nodes on each item arrival. Otherwise, to maintain the redundancy, we consider the following two update policies:

We now introduce a generic framework for CMS, that besides its normal activity, supports batching, as shown in Algo. 2. Then, we examine some batch representations in more details.

Define $B^{\prime }$ to count the delayed items between the consecutive shares, $B^{\prime }\le B$. Initially, as well as after sharing the batch, counter $B^{\prime }$ is zeroed and batch representation is reset too. When an item $x$ arrives, in addition to the normal CMS execution, counter $B^{\prime }$ is incremented and an item is processed into a batch, where $B^{\prime }$’s recent value can be used as an index, e.g., in Buffer representations.

The general scheme for send procedures is provided in Algo. 3. Representations that record counters need to implement $\mathrm{𝐸𝑛𝑐𝑜𝑑𝑒𝐹𝑖𝑥𝑒𝑑}$ and $\mathrm{𝐸𝑛𝑐𝑜𝑑𝑒𝑉𝑎𝑟}$ functions, while others need to implement $\mathrm{𝐸𝑛𝑐𝑜𝑑𝑒𝐼𝑡𝑒𝑚𝑠}$. When encoding rows of a fixed length, we can specify a row-width (actual number of elements) into a message’s header. However, when encoding variable-length rows, which is also the case when sketch is partitioned by cells, we need to encode these rows such that a redundant node can recognize the sketch row, each decoded element belongs to. For this purpose, besides the elements associated with a row, we first specify the number of elements to appear on that row. We call this metadata encoding.

We examine several data structures to economically represent and encode a batch of changes. We consider two main categories – item-based vs counter-based data structures, each of which can be implemented as a Buffer or a Compact Hash Table. Table 2 lists the main aspects of their complexity, e.g., the extra local space they require, the amount of traffic they produce on each cycle, and the extra computation they imply, while omitting $O$ notation for space saving. For further evaluation, these are expressed using hyper-parameters. We then compare the representations to each other, and to the baseline option of a full share, as well.

Buffer of Items.

The naïve approach for handling incremental shares is to record the arrived items in a local buffer, $\mathrm{𝐼𝑡𝑒𝑚𝐵𝑢𝑓𝑓}$. Upon filling the buffer with at most $B$ items during the cycle, simply send the content to all the covering redundant nodes, as defined by coverage mapping. A receiving redundant node will then extract the identifiers and execute an update procedure against its sum sketch, recalculating the hash functions for each delivered item.

Buffer of Items is a simple and easy-to-implement, but also has some drawbacks:

• $\mathrm{■}$

  Extra hash calculations must be performed by the redundant nodes on update.

• $\mathrm{■}$

  All the buffered items must be sent to every covering redundant node, $r_c\ge f$.

• $\mathrm{■}$

  No room for privacy, since items’ identifiers are openly shared.

• $\mathrm{■}$

  Different instances of a flow are handled independently of each other – duplicates.

To overcome these, we may ($i$) turn to counter-based representations, and/or ($ii$) avoid duplicates through aggregation.

Buffer of Counters.

Turning to counter-based representation, only a data node performs CMS hash calculations. The resulting indices are captured in a buffer of counters ${\mathrm{𝐶𝑛𝑡𝐵𝑢𝑓𝑓}}_{d\times B}$. For each counter, a data node knows exactly which $f$ redundant nodes cover it, so the bare minimum of $f$ copies can be sent. A receiving redundant node will then extract the indices and increment the corresponding counts of its sum sketch. As all the nodes are identically configured, we cannot guarantee that the privacy is protected with counters. Yet, it’s clearly better preserved than if sharing items.

Hash Table of Flows.

$\mathrm{𝐹𝑙𝑤𝐻𝑎𝑠ℎ}$ tries to improve $\mathrm{𝐼𝑡𝑒𝑚𝐵𝑢𝑓𝑓}$ through duplicate avoidance, but it requires longer elements to additionally store the corresponding frequencies, using key-value pairs, as well as additional capacity due to the load threshold $\alpha$ of a hash table.

Hash Table of Counters.

Similarly, $\mathrm{𝐶𝑛𝑡𝐵𝑢𝑓𝑓}$ can be replaced by $d$ hash tables of counters, where ${\mathrm{𝐶𝑛𝑡}}_i\mathrm{𝐻𝑎𝑠ℎ}$ is associated with a sketch row $i$.

In this paper, we evaluate the usage of compact hash tables through a comparison between $\mathrm{𝐹𝑙𝑤𝐻𝑎𝑠ℎ}$ and $\mathrm{𝐼𝑡𝑒𝑚𝐵𝑢𝑓𝑓}$. We generally prioritize to reduce the communication traffic and avoid early transmissions, but space efficiency is also desired. To evaluate the traffic produced by these representations, we need to compare $\frac{\lceil {\log }_{2}mid\rceil +\lceil {\log }_{2}B\rceil }{\lceil {\log }_{2}mid\rceil }=\theta$ vs $\frac{B}{b}=\beta$. We measured the traces using various batch sizes and hence, we consider the results as the range for any smaller Internet trace, e.g., within campus. The source code for measuring $\beta$ and generating the plots, as shown in Fig. 3 and Fig. 4, is now available at GitHub [6]. Given $\widehat{\beta }$ we further estimate the number of unique flows $\widehat{b}=\lceil \frac{B}{\widehat{\beta }}\rceil$, and derive the initial capacity for $\lceil \frac{\widehat{b}}{\alpha }\rceil$ buckets.