Scalable Analytics: Chapter 9

Characteristics of the Data Stream Model

In many modern data mining scenarios, the entire data set is not available in advance. This necessitates stream management, especially when the input rate is controlled by external sources rather than the system itself. Examples include Google search queries or status updates on platforms like Twitter and Facebook. Such data is characterized as being potentially infinite and non-stationary, meaning its statistical properties may change over time.

• Input Rate and Storage: Data elements enter the system at a rapid rate through one or more input ports, known as streams. Because the volume is so high, the system cannot store the entire stream in an accessible way (e.g., on disk or in main memory).

• Memory Constraints: The fundamental challenge is making critical calculations about the stream using a strictly limited amount of memory, typically secondary memory.

• General Stream Processing Architecture:

  • Processor: Handles the incoming elements (tuples).

  • Limited Working Storage: Used for immediate processing needs.

  • Archival Storage: For long-term data that does not need to be accessed quickly.

  • Query Types: The system must support both \text{Ad-Hoc Queries} (one-time requests) and \text{Standing Queries} (queries that run continuously as new data arrives).

Principal Stream Mining Problems and Applications

There are several core types of queries or operations performed on data streams:

• Sampling: Selecting a representative subset of the stream.

• Sliding Windows: Performing queries over only the most recent $N$ elements.

• Filtering: Identifying specific elements of interest from the stream.

Diverse Applications of Stream Mining

• Query Streams: Google monitors streams to identify queries that are becoming more frequent over time.

• Click Streams: Yahoo analyzes web traffic to detect unusual hits on specific pages within short intervals.

• Social Network News Feeds: Platforms like Twitter and Facebook track news feeds to identify trending topics.

• Sensor Networks: A central controller monitors data from numerous sensors reporting environmental or mechanical data.

• Telephone Call Records: Streams are used for customer billing and calculating settlements between different telecommunications providers.

• Network Monitoring: IP packets are monitored at switches to optimize routing protocols or to detect malicious activity, such as denial-of-service (DoS) attacks.

Sampling from a Data Stream

When it is impossible to store an entire stream, maintaining a sample is a primary strategy. There are two distinct sampling problems:

1. Fixed Proportion Sampling: Sampling a set fraction (e.g., $1/10$) of all incoming elements.

2. Fixed-Size Sampling: Maintaining a random sample of a specific size $s$ over a stream of unknown and potentially infinite length. This requires that at any time step $k$, each of the $k$ elements seen so far has an equal probability of being in the sample.

Problem 1: Sampling a Fixed Proportion

In a search engine scenario where tuples consist of $\text{(user, query, time)}$ and we have space for only $1/10$ of the stream, a naïve approach might fail.

• Failure of Naïve Sampling: If we store every query with a probability of $1/10$ (by generating a random integer from $0$ to $9$ and keeping only if the result is $0$), we cannot accurately answer questions about duplicate queries. For example, if a user issues $x$ queries once and $d$ queries twice (total queries = $x + 2d$), we would sample $x/10$ singletons and $2d/10$ duplicate queries at least once. However, the probability of sampling both queries in a duplicate pair is only $1/100$. The resulting sample-based estimate for the fraction of duplicates would be $\frac{d}{10x + 19d}$, which is incorrect.

• Generalized Solution (Sampling by Key): To maintain the integrity of user behavior, we pick a fixed proportion of users rather than queries. We choose a key (e.g., the user ID) and hash it into $b$ buckets. To sample a fraction $a/b$, we keep all tuples whose key hashes to a value at most $a$. This ensures that if a user is selected, all their queries are sampled.

Problem 2: Maintaining a Fixed-Size Sample (Reservoir Sampling)

To maintain a sample $S$ of exactly size $s$ when the total number of items $n$ is unknown, we use the \text{Reservoir Sampling} algorithm.

• Algorithm Steps:

  1. Store the first $s$ elements of the stream directly into the sample $S$.

  2. When the $n$-th element arrives (n > s), keep it with probability $\frac{s}{n}$. If it is not kept, it is discarded.

  3. If the $n$-th element is kept, it replaces one of the existing $s$ elements in $S$, chosen uniformly at random.

• Proof by Induction:

  • Base Case: After $n = s$ elements, every element is in the sample with probability $s/s = 1$.

  • Inductive Step: Assume after $n$ elements, each has a probability of $s/n$ of being in the sample. For the $(n+1)$-th element, the probability it is included is $s/(n+1)$. For elements already in $S$, the probability they remain in the sample is calculated as:         $\text{P(Remaining)} = \text{P(Item } n+1 \text{ discarded)} + \text{P(Item } n+1 \text{ kept} \times \text{Existing item not replaced)}$         $\text{P(Remaining)} = (1 - \frac{s}{n+1}) + (\frac{s}{n+1} \times \frac{s-1}{s})$         $\text{P(Remaining)} = \frac{n+1-s}{n+1} + \frac{s-1}{n+1} = \frac{n}{n+1}$

  • The total probability that a previous element is in the sample at time $n+1$ is: $\frac{s}{n} \times \frac{n}{n+1} = \frac{s}{n+1}$.

Queries over a Sliding Window

Sliding windows focus on the most recent $N$ elements. This is useful for tracking recent trends (e.g., Amazon sales of product $X$ in the last $k$ transactions). If $N$ is extremely large (e.g., $1$ billion bits), storing the window explicitly is impossible.

The Counting Bits Problem

We need to estimate the number of $1\text{s}$ in the last $k$ bits of a stream where $k \le N$. A simple solution using uniformity assumes the fraction of $1\text{s}$ in the last $N$ bits is equal to the total fraction of $1\text{s}$ seen so far ($N \times \frac{S}{S+Z}$, where $S$ is the count of $1\text{s}$ and $Z$ is the count of $0\text{s}$), but this fails if the stream is non-uniform.

The DGIM Method (Datar, Gionis, Indyk, Motwani)

This method provides an approximate answer with an error bound of no more than $50\%$, using only $O(\log^2 N)$ bits of storage per stream.

• Timestamps: Each bit is assigned a timestamp modulo $N$, requiring $O(\log N)$ bits.

• Buckets: A bucket consists of the timestamp of its end and the number of $1\text{s}$ between its start and end. The number of $1\text{s}$ (the size) must be a power of $2$.

• Bucket Rules:

  • There can be either one or two buckets for each power-of-$2$ size.

  • Buckets do not overlap in time and are sorted by size.

  • A bucket is deleted when its end-time is more than $N$ units in the past.

• Updating Buckets: When a new bit arrives:

  1. If the bit is $0$, no changes are made (except potential deletions of old buckets).

  2. If the bit is $1$, create a new bucket of size $1$. If this results in three buckets of size $1$, combine the oldest two into one bucket of size $2$. If this creates three buckets of size $2$, combine the oldest two into a bucket of size $4$, and so on down the line.

• Querying: To estimate the number of $1\text{s}$ in the last $N$ bits, sum the sizes of all buckets except the oldest one, and add half the size of the oldest bucket.

• Error Proof: If the last bucket has size $2^r$, an error of at most $2^{r-1}$ is made by assuming half its bits are in the window. Since there is at least one bucket of every size smaller than $2^r$, the total sum is at least $1 + 2 + 4 + \dots + 2^{r-1} = 2^r - 1$. Thus, the error relative to the total is at most $50\%$. This error can be reduced to $O(1/r)$ by allowing $r-1$ or $r$ buckets of each size.

Filtering Data Streams and Bloom Filters

Filtering involves checking if stream elements belong to a predefined set of keys $S$. When $S$ is too large for a standard hash table (e.g., $1$ billion email addresses in a spam filter), probabilistic methods are used. These methods ensure no false negatives but allow for some false positives.

First Cut Solution Analysis (Throwing Darts)

Consider throwing $m$ darts (hash values of items in $S$) into $n$ targets (bits in a bit array $B$).

• The probability a target is not hit by any dart is approximately $(1 - 1/n)^m \approx e^{-m/n}$.

• The probability a bit is set to $1$ (the false positive rate) is $1 - e^{-m/n}$.

• Example: With $10^9$ email addresses and an $8\,GB$ bit array ($8 \times 10^9$ bits), the false positive rate is $1 - e^{-1/8} = 0.1175$.

Bloom Filters

A Bloom Filter uses $k$ independent hash functions $h_1, \dots, h_k$.

• Initialization: Set all bits in array $B$ to $0$. For every member $s \in S$, set $B[h_i(s)] = 1$ for all $i = 1, \dots, k$.

• Runtime: For a stream element $a$, if $B[h_i(a)] = 1$ for all $i$, declare that $a$ is in $S$. Otherwise, discard it.

• Bloom Filter Analysis:

  • The fraction of bits set to $1$ is $(1 - e^{-km/n})$.

  • The false positive probability is $(1 - e^{-km/n})^k$.

  • The optimal number of hash functions to minimize false positives is $k = \frac{n}{m} \ln(2)$. For a ratio of $n/m = 8$, the optimal $k$ is approximately $6$.

• Bloom Filters are efficient and suitable for hardware implementation due to their low memory requirements.