Comprehensive Guide to Stream Processing, Complex Event Processing, and Stream Mining

Stream Mining Fundamentals and the Heavy Hitters Problem

• Heavy Hitters Definition:

  • In a data stream of total length $m$, the heavy hitters problem consists of identifying all items $i$ whose arrival frequency $f_i$ exceeds a predefined threshold fraction of the total arrivals:
    $f_i > \frac{m}{k}$     where $k$ is a user-defined integer parameter.
  • For example, in an arrival stream of $m = 15$ items: [1, 1, 2, 3, 4, 5, 1, 1, 1, 5, 3, 3, 1, 1, 2] with $k = 3$, the objective is to find all elements that occur strictly more than $\frac{15}{3} = 5$ times. In this sequence, item 1 appears 7 times and represents the sole heavy hitter.

• Resource Bounds and Theoretical Guarantees:

  • Space usage is strictly bounded by:
    $O(k \cdot (\log(m) + \log(n)))$     where $n$ represents the size of the item universe and $m$ represents the stream sequence length.
  • The algorithm guarantees a one-sided error: no false negatives occur (all true heavy hitters exceeding $\frac{m}{k}$ are guaranteed to be returned), but false positives may occur (infrequent elements may occasionally remain in the summary counters at stream termination).

• The Misra-Gries Algorithm:

  • Maintains an associative array (or dictionary) $D$ containing at most $k - 1$ candidate keys along with their corresponding positive integer counters.
  • For each arriving element $e_j$ from the stream:
  • If $e_j \in \text{keys}(D)$, increment its counter:
    $D[e_j] \leftarrow D[e_j] + 1$
  • Else if the number of monitored keys satisfies $|\text{keys}(D)| < k - 1$, allocate a new entry:
    $D[e_j] \leftarrow 1$
  • Else (when the dictionary is completely full and $e_j$ is not present), decrement the counter of every monitored key by 1:
    $\forall \ell \in \text{keys}(D): D[\ell] \leftarrow D[\ell] - 1$       Following decrement, immediately evict all keys whose counter values reach 0.
  • At stream termination, output estimated frequencies $\hat{f}_i = D[i]$ for all $i \in \text{keys}(D)$, and set $\hat{f}_i = 0$ for all unmonitored items $i \notin \text{keys}(D)$.

• Execution Trace:
  • Consider tracking heavy hitters with capacity $k - 1 = 3$ across the arrival sequence: A, B, C, D, A, B, E, A, B, C, A.

• Step 1: Add A \rightarrow \{A: 1\}$.\n * Step 2: Add `B` \rightarrow {A: 1, B: 1}$.

• Step 3: Add C \rightarrow \{A: 1, B: 1, C: 1\}$ (map reaches capacity of 3).\n * Step 4: Arrive `D`. Map is full; decrement all entries: `A` becomes 0, `B` becomes 0, `C` becomes 0. Evict all \rightarrow {}$.

• Step 5: Add A \rightarrow \{A: 1\}$.\n * Step 6: Add `B` \rightarrow {A: 1, B: 1}$.

• Step 7: Add E \rightarrow \{A: 1, B: 1, E: 1\}$.\n * Step 8: Increment `A` \rightarrow {A: 2, B: 1, E: 1}$.

• Step 9: Increment B \rightarrow \{A: 2, B: 2, E: 1\}$.\n * Step 10: Arrive `C`. Map is full; decrement all: `A` becomes 1, `B` becomes 1, `E` becomes 0 (evicted) \rightarrow {A: 1, B: 1}$. Add C \rightarrow \{A: 1, B: 1, C: 1\}$.\n * Step 11: Increment `A` \rightarrow {A: 2, B: 1, C: 1}$.

• Final output retains heavy hitters A (count 2), B (count 1), and C (count 1).

  • Algorithmic Edge Cases:

• When all counter slots are full and an unmonitored element arrives, every counter is decremented simultaneously. If all counters were strictly greater than 1, no counts reach zero, and the new arriving element is discarded without any key eviction.

• Adversarial or pathological input streams where high-frequency items alternate with bursts of distinct items can systematically drive candidate counters down to zero, causing true heavy hitters to be repeatedly evicted and reported with severe undercounts.

Frequent Pattern Mining and the Lossy Counting Algorithm

• Challenges of Frequent Itemset Mining on Streams:

  • One-Scan Constraint: Data arrives continuously and cannot be re-read from secondary storage.
  • Memory Bounds: Working memory is strictly limited relative to the infinite stream volume.
  • Combinatorial Explosion: The number of potential candidate itemsets grows exponentially with transaction width.
  • Frequency Inversion (Concept Drift): Infrequent itemsets from earlier windows may suddenly surge and become frequent, while historically dominant itemsets may decay into insignificance.

• Methodological Paradigms:

  • Landmark Windows: Treat all historical items uniformly from an initial fixed point in time without decaying past data.
  • Sliding Windows and Decay Factors: Emphasize recent observations by discarding items older than a temporal threshold or applying exponential decay weights.
  • Multi-Granularity Time Windows: Maintain summary statistics across multiple temporal granularities (e.g., minutes, hours, days).

• The Lossy Counting Algorithm:

  • A deterministic, one-pass stream algorithm designed to compute frequency counts over an unbounded sequence.
  • Accepts two user-specified parameters: a support threshold $\sigma$ and an error tolerance parameter $\epsilon$.
  • The incoming stream is conceptually partitioned into consecutive "buckets" of width:
    $W = \frac{1}{\epsilon}$
  • The current stream length is denoted by $n$. The current bucket identifier is defined by:
    $\Delta = \left\lceil \frac{n}{W} \right\rceil = \lceil n \cdot \epsilon \rceil$
  • Maintains a set of monitored tuples $T$, where each entry stores the item identifier $e$, its approximate frequency count $\text{Count}_e$, and the maximum error associated with its insertion $\Delta_e$.

• Lossy Counting Step-by-Step Procedure:

  • For each arriving element $e$:
  1. Increment total processed stream length: $n \leftarrow n + 1$.
  2. If $e$ is already monitored ($e \in T$), increment its counter:
     $\text{Count}_e \leftarrow \text{Count}_e + 1$
  3. If $e$ is not monitored ($e \notin T$), insert it with an initial count and current bucket offset:
     $T \leftarrow T \cup \{ (e, 1 + \Delta) \}$
  4. Evaluate whether a bucket boundary has been crossed by testing if $\lceil n \cdot \epsilon \rceil > \Delta$.
  5. Whenever a bucket boundary transition occurs, increment $\Delta \leftarrow \Delta + 1$ and execute a pruning pass across all monitored elements $j \in T$:
     • If $\text{Count}_j < \Delta$, delete $j$ from $T$.

• Trace Example ($\epsilon = 0.5$, Bucket Width $W = \frac{1}{0.5} = 2$, Stream: [A, B, A, C, A]):

  • Initial State: $n = 0$, $\Delta = 0$, $T = \emptyset$.
  • Step 1 ($e = A$):
    $n = 1$. Item A is not in $T$. Add A to $T$ with count 1: $T = \{A\}$, $\text{Count}_A = 1$.
        Threshold evaluation: $\lceil \frac{1}{2} \rceil = 1$. Since $1 > \Delta$ ($1 > 0$), enter cleanup: update $\Delta \leftarrow 1$.
        Prune check: For A, is $\text{Count}_A < \Delta$ ($1 < 1$)? False. Retain A.
    End of Step 1 State: $T = \{A\}$, $\text{Count}_A = 1$, $\Delta = 1$.
  • Step 2 ($e = B$):
    $n = 2$. Item B is not in $T$. Add B to $T$: $T = \{A, B\}$, $\text{Count}_B = 1$.
        Threshold evaluation: $\lceil \frac{2}{2} \rceil = 1$. Since $1 = \Delta$ ($1 = 1$), skip cleanup.
    End of Step 2 State: $T = \{A, B\}$, $\text{Count}_A = 1$, $\text{Count}_B = 1$, $\Delta = 1$.
  • Step 3 ($e = A$):
    $n = 3$. Item A is in $T$. Increment count: $\text{Count}_A = 1 + 1 = 2$.
        Threshold evaluation: $\lceil \frac{3}{2} \rceil = 2$. Since $2 > \Delta$ ($2 > 1$), enter cleanup: update $\Delta \leftarrow 2$.
        Prune check: For A, $\text{Count}_A = 2 < 2$ is False (keep A). For B, $\text{Count}_B = 1 < 2$ is True (delete B).
    End of Step 3 State: $T = \{A\}$, $\text{Count}_A = 2$, $\Delta = 2$.
  • Step 4 ($e = C$):
    $n = 4$. Item C is not in $T$. Add C to $T$: $T = \{A, C\}$, $\text{Count}_C = 1$.
        Threshold evaluation: $\lceil \frac{4}{2} \rceil = 2$. Since $2 = \Delta$ ($2 = 2$), skip cleanup.
    End of Step 4 State: $T = \{A, C\}$, $\text{Count}_A = 2$, $\text{Count}_C = 1$, $\Delta = 2$.
  • Step 5 ($e = A$):
    $n = 5$. Item A is in $T$. Increment count: $\text{Count}_A = 2 + 1 = 3$.
        Threshold evaluation: $\lceil \frac{5}{2} \rceil = 3$. Since $3 > \Delta$ ($3 > 2$), enter cleanup: update $\Delta \leftarrow 3$.
        Prune check: For A, $\text{Count}_A = 3 < 3$ is False (keep A). For C, $\text{Count}_C = 1 < 3$ is True (delete C).
    End of Step 5 State: $T = \{A\}$, $\text{Count}_A = 3$, $\Delta = 3$.

• Practice Scenarios for Lossy Counting Execution:

  • Scenario 1 (Search Queries): $\epsilon = 0.25$, bucket width $W = \frac{1}{0.25} = 4$, stream sequence [X, Y, X, X, Z].
  • Step 1 ($n = 1, e = X$): $X$ inserted ($\text{Count}_X = 1$), $\lceil 1/4 \rceil = 1 > 0 \implies \Delta = 1$, $X$ kept ($1 \not< 1$).
  • Step 2 ($n = 2, e = Y$): $Y$ inserted ($\text{Count}_Y = 1$), $\lceil 2/4 \rceil = 1 = \Delta$, no prune.
  • Step 3 ($n = 3, e = X$): $\text{Count}_X = 2$, $\lceil 3/4 \rceil = 1 = \Delta$, no prune.
  • Step 4 ($n = 4, e = X$): $\text{Count}_X = 3$, $\lceil 4/4 \rceil = 1 = \Delta$, no prune.
  • Step 5 ($n = 5, e = Z$): $Z$ inserted ($\text{Count}_Z = 1$), $\lceil 5/4 \rceil = 2 > 1 \implies \Delta = 2$. Prune check: $\text{Count}_X = 3 \ge 2$ (kept); $\text{Count}_Y = 1 < 2$ (deleted); $\text{Count}_Z = 1 < 2$ (deleted). Final monitored set: $T = \{X\}$.

• Scenario 2 (Packet Headers): $\epsilon = 0.2$, bucket width $W = \frac{1}{0.2} = 5$, stream sequence [A, B, A, C, B, A].

  • Steps 1 to 5 process items within bucket 1 ($\Delta = 1$). At step 5, $\lceil 5/5 \rceil = 1 = \Delta$, no cleanup.
  • At step 6 ($e = A, n = 6$), $\lceil 6/5 \rceil = 2 > 1 \implies \Delta = 2$. Cleanup runs, evicting any monitored item with count strictly less than 2.

Stream Clustering: Fundamentals, Leader Clustering, and Single-Pass k-Means

• Cluster Feature (CF) Representation:

  • Also called a micro-cluster, a Cluster Feature is a compact summary structure defined as an ordered triple:
    $\mathbf{CF} = (N, \mathbf{LS}, \mathbf{SS})$
  • Components:
  • $N$: Total number of data points summarized in the cluster.
  • $\mathbf{LS}$: Linear sum vector of the $N$ points across all dimensions:
    $\mathbf{LS} = \sum_{i=1}^N \mathbf{x}_i$
  • $\mathbf{SS}$: Square sum vector storing the coordinate-wise squared sums:
    $\mathbf{SS} = \sum_{i=1}^N \mathbf{x}_i^2$
  • Worked Example (2D Points $(1, 2)$ and $(3, 4)$):
  • $N = 2$
  • $\mathbf{LS} = (1 + 3, 2 + 4) = (4, 6)$
  • $\mathbf{SS} = (1^2 + 3^2, 2^2 + 4^2) = (1 + 9, 4 + 16) = (10, 20)$

• Mathematical Properties of Cluster Features:

  • Incrementality: When an individual point $\mathbf{x}$ arrives, the cluster summary updates in $O(1)$ time without reprocessing past points:
    $N \leftarrow N + 1$$\mathbf{LS} \leftarrow \mathbf{LS} + \mathbf{x}$$\mathbf{SS} \leftarrow \mathbf{SS} + \mathbf{x}^2$
  • Additivity: If two disjoint clusters $A$ and $B$ are merged into cluster $C$, their sufficient statistics sum directly:
    $N_C = N_A + N_B$$\mathbf{LS}_C = \mathbf{LS}_A + \mathbf{LS}_B$$\mathbf{SS}_C = \mathbf{SS}_A + \mathbf{SS}_B$

• Derived Cluster Characterization Measures:

  • Centroid ($\mathbf{X}_0$): Represents the gravity center of the cluster:
    $\mathbf{X}_0 = \frac{\mathbf{LS}}{N}$
  • Radius ($R$): Average distance from member points to the centroid:
    $R = \sqrt{\frac{\sum_{i=1}^N (\mathbf{x}_i - \mathbf{X}_0)^2}{N}} = \sqrt{\frac{\mathbf{SS} - \frac{\mathbf{LS}^2}{N}}{N}}$
  • Diameter ($D$): Average pairwise distance between all member points:
    $D = \sqrt{\frac{\sum_{i=1}^N \sum_{j=1}^N (\mathbf{x}_i - \mathbf{x}_j)^2}{N(N - 1)}} = \sqrt{\frac{2N \cdot \mathbf{SS} - 2 \cdot \mathbf{LS}^2}{N(N - 1)}}$
  • Distance norms between points or centroids can be computed using the $L_1$ (Manhattan) norm:
    $L_1 = \sum_{i=1}^n |x_{a_i} - x_{b_i}|$     or the $L_2$ (Euclidean) norm:
    $L_2 = \sqrt{\sum_{i=1}^n (x_{a_i} - x_{b_i})^2}$

• The Leader Clustering Algorithm:

  • A single-pass, incremental clustering technique governed by a user-defined control distance threshold $\delta$ (or $T$).
  • Procedure:
  1. Initialize the first point $x_1$ as the centroid of the first cluster: $C = \{x_1\}$.
  2. For each incoming point $x_i$, identify the closest existing leader centroid $c_r \in C$.
  3. If the distance satisfies $d(x_i, c_r) < \delta$, assign $x_i$ to cluster $c_r$.
  4. If $d(x_i, c_r) \ge \delta$, instantiate a new cluster with $x_i$ as its leader: $C \leftarrow C \cup \{x_i\}$.
  • Evaluation:
  • Advantages: Requires only one pass over the data stream; fast execution; does not require fixing the number of clusters $k$ in advance.
  • Limitations: Highly sensitive to arrival order; arbitrary ordering yields entirely different cluster boundaries. Threshold selection is brittle—small thresholds cause cluster proliferation, while large thresholds merge distinct natural groupings.
  • Worked Example: Stream $X = [2, 3, 8, 9, 15]$ using absolute distance and $\delta = 3$.

* Point `2`: First point, establishes Leader 2 ($\{2\}$).
* Point `3`: Closest leader is 2; distance $|3 - 2| = 1 < 3$. Joins Leader 2 ($\{2, 3\}$).
* Point `8`: Closest leader is 2; distance $|8 - 2| = 6 \ge 3$. Creates new cluster with Leader 8 ($\{8\}$).
* Point `9`: Closest leader is 8; distance $|9 - 8| = 1 < 3$. Joins Leader 8 ($\{8, 9\}$).
* Point `15`: Closest leader is 8; distance $|15 - 8| = 7 \ge 3$. Creates new cluster with Leader 15 ($\{15\}$).
* Resulting Clusters: Leader 2: $\left\{2, 3\right\}$, Leader 8: $\left\{8, 9\right\}$, Leader 15: $\left\{15\right\}$.

• Practice Problem: Sequence $S = \{5, 7, 12, 13, 22\}$ with threshold $T = 4$:

  • Point 5: Leader 5 ($\{5\}$).

  • Point 7: $|7 - 5| = 2 < 4 \implies$ Joins Leader 5 ($\{5, 7\}$).

  • Point 12: $|12 - 5| = 7 \ge 4 \implies$ Establishes Leader 12 ($\{12\}$).

  • Point 13: $|13 - 12| = 1 < 4 \implies$ Joins Leader 12 ($\{12, 13\}$).

  • Point 22: $|22 - 12| = 10 \ge 4 \implies$ Establishes Leader 22 ($\{22\}$).

  • Single-Pass k-Means Clustering:

• Addresses the challenge of clustering massive streams where standard k-means (which requires repeatedly loading the full dataset across multiple iterative passes) cannot execute.

• Operates across successive data blocks loaded into a fixed-size buffer while maintaining sufficient statistics in a discard set for each cluster.

• Workflow:

  1. Randomly initialize $k$ cluster means.
  2. Initialize empty discard sets (Count $N_j = 0$, Sum $\text{SUM}_j = 0$) for each cluster.
  3. While the stream is active, fill the buffer with incoming raw points.
  4. Run standard k-means iterations on points and discard set entries until convergence. Each discard set entry is treated as a single point placed at its centroid, weighted by its count $N_j$.
  5. For each group, update the sufficient statistics ($N$ and $\text{SUM}$) of the discard set with the new examples assigned to that group.
  6. Clear individual raw points from the buffer and repeat with the next block.

• Detailed Trace 1 (Buffer 1):

  • Given: $k = 2$, initial centroids $\mu_1 = 2.0$, $\mu_2 = 8.0$. Discard statistics: $N_1 = 0$, $\text{SUM}_1 = 0$; $N_2 = 0$, $\text{SUM}_2 = 0$.
  • Buffer contents: $S_{\text{buffer}} = \{3.0, 4.0, 9.0\}$.
  • Point 3.0: Distance to $\mu_1 = |3.0 - 2.0| = 1.0$; distance to $\mu_2 = |3.0 - 8.0| = 5.0$. Assign to Cluster 1.
  • Point 4.0: Distance to $\mu_1 = |4.0 - 2.0| = 2.0$; distance to $\mu_2 = |4.0 - 8.0| = 4.0$. Assign to Cluster 1.
  • Point 9.0: Distance to $\mu_1 = |9.0 - 2.0| = 7.0$; distance to $\mu_2 = |9.0 - 8.0| = 1.0$. Assign to Cluster 2.
  • Updated Statistics after Buffer 1:
  • Cluster 1: Total Count $N_1 = 2$, Total Sum $\text{SUM}_1 = 3.0 + 4.0 = 7.0$, Updated Centroid \mu_1 = \frac{7.0}{2} = 3.5$.\n * Cluster 2: Total Count N_2 = 1$, Total Sum$\text{SUM}_2 = 9.0$, Updated Centroid$\mu_2 = \frac{9.0}{1} = 9.0$.

• Detailed Trace 2 (Buffer 2):

  • Incoming points: $S_{\text{buffer}} = \{2.0, 5.0, 15.0\}$. Current centroids: $\mu_1 = 3.5$, $\mu_2 = 9.0$.
  • Point 2.0: Distance to $\mu_1 = |2.0 - 3.5| = 1.5$; distance to $\mu_2 = |2.0 - 9.0| = 7.0$. Assign to Cluster 1.
  • Point 5.0: Distance to $\mu_1 = |5.0 - 3.5| = 1.5$; distance to $\mu_2 = |5.0 - 9.0| = 4.0$. Assign to Cluster 1.
  • Point 15.0: Distance to $\mu_1 = |15.0 - 3.5| = 11.5$; distance to $\mu_2 = |15.0 - 9.0| = 6.0$. Assign to Cluster 2.
  • Cumulative Sufficient Statistics:
  • Cluster 1: $N_1 = 2 + 2 = 4$, $\text{SUM}_1 = 7.0 + (2.0 + 5.0) = 14.0$, Updated Centroid \mu_1 = \frac{14.0}{4} = 3.5$.\n * Cluster 2: N_2 = 1 + 1 = 2$,$\text{SUM}_2 = 9.0 + 15.0 = 24.0$, Updated Centroid$\mu_2 = \frac{24.0}{2} = 12.0$.

• Practice Exercise 2 Formulation:

  • Cluster 1 initial status: points $(1, 2, 3)$, $N_1 = 3$, $\text{LS}_1 = 6$, Centroid \mu_1 = \frac{6}{3} = 2.0$.\n * Cluster 2 initial status: points (10, 11, 12)$,$N_2 = 3$,$\text{LS}_2 = 33$, Centroid$\mu_2 = \frac{33}{3} = 11.0$.
  • Incoming points are assigned against $\mu_1 = 2.0$ and $\mu_2 = 11.0$, accumulating directly into $(N, \text{LS})$ before recalculating centroids.

BIRCH (Balanced Iterative Reducing and Clustering using Hierarchies)

• Architecture and Operational Phases:

  • Designed specifically for clustering large-scale and streaming numeric data.
  • Operates across two principal phases:
  1. Phase 1 (CF-Tree Construction): Scans the input stream sequentially and builds an in-memory, height-balanced Clustering Feature Tree (CF-Tree) that dynamically compresses dense regions into micro-clusters while filtering outliers.
  2. Phase 2 (Global Clustering): Applies an arbitrary clustering algorithm (such as agglomerative hierarchical clustering or k-means) directly to the leaf summaries at the base of the tree.

• The CF-Tree Structural Hierarchy:

  • A CF-Tree is a balanced search tree characterized by two fundamental parameters:
  • Branching Factor ($B$): The maximum number of child entries allowed in any non-leaf or leaf node.
  • Threshold ($T$): The maximum allowable cluster radius (or diameter) for any leaf node micro-cluster.
  • Node Structure:
  • Non-Leaf Nodes: Contain at most $B$ entries of the form $[\mathbf{CF}_i, \text{child}_i]$, where $\text{child}_i$ points to a lower-level subtree and $\mathbf{CF}_i$ stores the aggregate sufficient statistics of all entries in that subtree.
  • Leaf Nodes: Contain at most $B$ (or $L$) leaf entries, each storing an individual micro-cluster $\mathbf{CF} = (N, \mathbf{LS}, \mathbf{SS})$.

• Point Insertion Procedure:

  1. Identify Closest Leaf: Starting at the root, recursively traverse down the tree by selecting the non-leaf entry whose centroid is closest to the arriving data point $\mathbf{x}$.
  2. Test Threshold Constraint: At the target leaf, identify the closest leaf entry $\mathbf{CF}_k$. Compute the tentative radius $R_{\text{new}}$ if $\mathbf{x}$ were merged into $\mathbf{CF}_k$:      $R_{\text{new}} = \sqrt{\frac{(\mathbf{SS} + \mathbf{x}^2) - \frac{(\mathbf{LS} + \mathbf{x})^2}{N + 1}}{N + 1}}$
  3. Absorb or Allocate:
  • If $R_{\text{new}} \le T$, absorb $\mathbf{x}$ into $\mathbf{CF}_k$ and update its triplet:
    $N \leftarrow N + 1, \quad \mathbf{LS} \leftarrow \mathbf{LS} + \mathbf{x}, \quad \mathbf{SS} \leftarrow \mathbf{SS} + \mathbf{x}^2$
  • If $R_{\text{new}} > T$, create a new leaf entry containing $\mathbf{x}$.
  1. Node Splitting: If creating a new entry causes the leaf node to exceed capacity $B$, split the leaf into two distinct nodes. Choose the two entries with the maximum pairwise distance as seeds, reassign the remaining entries to the closest seed, and propagate new CF entries up to the parent. If the parent exceeds $B$, split recursively up to the root.

• Comprehensive Stepped Trace ($T < 1.5$, Branching Factor $B = 2$):

  • Given stream of ten 2D points:
    $X_1=(3,4), X_2=(2,6), X_3=(4,5), X_4=(4,7), X_5=(3,8), X_6=(6,2), X_7=(7,2), X_8=(7,4), X_9=(8,4), X_{10}=(7,9)$.

• Point $X_1 = (3, 4)$:

  • Single point in tree; radius $R = 0 < 1.5$.
  • Construct Leaf with entry \mathbf{CF}_1$: \n    N = 1, \quad \mathbf{LS} = (3, 4), \quad \mathbf{SS} = (3^2, 4^2) = (9, 16)\n\n* **Point X_2 = (2, 6):**\n * Tentative merger with \mathbf{CF}1$($N = 2):\n    \mathbf{LS} = (3, 4) + (2, 6) = (5, 10)\n    \mathbf{SS} = (9 + 2^2, 16 + 6^2) = (9 + 4, 16 + 36) = (13, 52)\n    R = \sqrt{\frac{(13, 52) - \frac{(5, 10)^2}{2}}{2}} = \sqrt{\frac{(13, 52) - (12.5, 50)}{2}} = \sqrt{(0.25, 1)} = (0.5, 1.0)\n * Since (0.5, 1.0) < (1.5, 1.5)$,$X_2$clusters into$\mathbf{CF}_1.\n * \mathbf{CF}_1 = \langle 2, (5, 10), (13, 52) \rangle.\n\n\n\n* **Point X_3 = (4, 5):**\n * Tentative merger with \mathbf{CF}_1$($N = 3):\n    \mathbf{LS} = (5, 10) + (4, 5) = (9, 15)\n    \mathbf{SS} = (13 + 4^2, 52 + 5^2) = (13 + 16, 52 + 25) = (29, 77)\n    R = \sqrt{\frac{(29, 77) - \frac{(9, 15)^2}{3}}{3}} = \sqrt{\frac{(29, 77) - (27, 75)}{3}} = \sqrt{\left(\frac{2}{3}, \frac{2}{3}\right)} \approx (0.47, 0.471)\n * Since (0.47, 0.471) < (1.5, 1.5)$,$X_3$clusters into$\mathbf{CF}_1.\n * \mathbf{CF}_1 = \langle 3, (9, 15), (29, 77) \rangle.\n\n* **Point X_4 = (4, 7):**\n * Tentative merger with \mathbf{CF}_1$($N = 4):\n    \mathbf{LS} = (9, 15) + (4, 7) = (13, 22)\n    \mathbf{SS} = (29 + 16, 77 + 49) = (45, 126)\n    R = \sqrt{\frac{(45, 126) - \frac{(13, 22)^2}{4}}{4}} = \sqrt{\frac{(45, 126) - (42.25, 121)}{4}} = (0.41, 0.55)\n * Since (0.41, 0.55) < (1.5, 1.5)$,$X_4$clusters into$\mathbf{CF}_1.\n * \mathbf{CF}_1 = \langle 4, (13, 22), (45, 126) \rangle.\n\n\n\n* **Point X_5 = (3, 8):**\n * Tentative merger with \mathbf{CF}_1$($N = 5):\n    \mathbf{LS} = (13, 22) + (3, 8) = (16, 30)\n    \mathbf{SS} = (45 + 9, 126 + 64) = (54, 190)\n    R = \sqrt{\frac{(54, 190) - \frac{(16, 30)^2}{5}}{5}} = \sqrt{\frac{(54, 190) - (51.2, 180)}{5}} = (0.33, 0.63)\n * Since (0.33, 0.63) < (1.5, 1.5)$,$X_5$clusters into$\mathbf{CF}_1.\n * \mathbf{CF}_1 = \langle 5, (16, 30), (54, 190) \rangle.\n\n* **Point X_6 = (6, 2):**\n * Tentative merger with \mathbf{CF}_1$($N = 6):\n    \mathbf{LS} = (16, 30) + (6, 2) = (22, 32)\n    \mathbf{SS} = (54 + 36, 190 + 4) = (90, 194)\n    R = \sqrt{\frac{(90, 194) - \frac{(22, 32)^2}{6}}{6}} = (1.24, 1.97)\n * Threshold breach: the y-radius 1.97 > 1.5$.$X_6$cannot merge into$\mathbf{CF}_1.\n * Allocate a new entry \mathbf{CF}_2$within the node (allowed since$B = 2):\n    \mathbf{CF}_2 = \langle 1, (6, 2), (36, 4) \rangle\n\n\n\n* **Point X_7 = (7, 2):**\n * Centroids: \mu(\mathbf{CF}_1) = \frac{(16, 30)}{5} = (3.2, 6.0)$;$\mu(\mathbf{CF}_2) = \frac{(6, 2)}{1} = (6.0, 2.0).\n * Distance to \mathbf{CF}_2$is substantially closer than to$\mathbf{CF}_1$. Test merger with$\mathbf{CF}_2$($N = 2):\n    \mathbf{LS} = (6, 2) + (7, 2) = (13, 4)\n    \mathbf{SS} = (36 + 49, 4 + 4) = (85, 8)\n    R = \sqrt{\frac{(85, 8) - \frac{(13, 4)^2}{2}}{2}} = (0.5, 0.0) < (1.5, 1.5)\n * Merges into \mathbf{CF}_2 = \langle 2, (13, 4), (85, 8) \rangle.\n\n* **Point X_8 = (7, 4):**\n * Centroids: \mu(\mathbf{CF}_1) = (3.2, 6.0)$;$\mu(\mathbf{CF}_2) = \frac{(13, 4)}{2} = (6.5, 2.0).\n * Closer to \mathbf{CF}_2$. Test merger with$\mathbf{CF}_2$($N = 3):\n    \mathbf{LS} = (13, 4) + (7, 4) = (20, 8)\n    \mathbf{SS} = (85 + 49, 8 + 16) = (134, 24)\n    R = \sqrt{\frac{(134, 24) - \frac{(20, 8)^2}{3}}{3}} = (0.47, 0.94) < (1.5, 1.5)\n * Merges into \mathbf{CF}_2 = \langle 3, (20, 8), (134, 24) \rangle.\n\n\n\n * **Point X_9 = (8, 4):**\n * Centroids: \mu(\mathbf{CF}_1) = (3.2, 6.0)$;$\mu(\mathbf{CF}_2) = \frac{(20, 8)}{3} = (6.67, 2.67).\n * Closer to \mathbf{CF}_2$. Test merger with$\mathbf{CF}_2$($N = 4):\n      \mathbf{LS} = (20, 8) + (8, 4) = (28, 12)\n      \mathbf{SS} = (134 + 64, 24 + 16) = (198, 40)\n      R = \sqrt{\frac{(198, 40) - \frac{(28, 12)^2}{4}}{4}} = (0.70, 1.0) < (1.5, 1.5)\n * Merges into \mathbf{CF}_2 = \langle 4, (28, 12), (198, 40) \rangle.\n\n * **Point X{10} = (7, 9):**\n * Centroids: \mu(\mathbf{CF}1) = (3.2, 6.0)$;$\mu(\mathbf{CF}_2) = \frac{(28, 12)}{4} = (7.0, 3.0).\n * Distance to \mathbf{CF}_1$is closer. Test merger with$\mathbf{CF}_1$($N = 6):\n      \mathbf{LS} = (16, 30) + (7, 9) = (23, 39)\n      \mathbf{SS} = (54 + 49, 190 + 81) = (103, 271)\n      R = \sqrt{\frac{(103, 271) - \frac{(23, 39)^2}{6}}{6}} = (1.57, 1.70)\n * Threshold breach: (1.57, 1.70) > (1.5, 1.5)$. Cannot merge with$\mathbf{CF}_1.\n * Testing merger with \mathbf{CF}_2 also breaches threshold.\n * Creating a new entry \mathbf{CF}_3 = \langle 1, (7, 9), (49, 81) \rangle$causes the node entry count to reach 3, violating the branching factor$B = 2.\n * **Node Split:** The node splits into two nodes. A new parent root node is formed containing two summarized branch entries:\n * Left Leaf: contains \mathbf{CF}_1 = \langle 5, (16, 30), (54, 190) \rangle.\n * Right Leaf: contains \mathbf{CF}_2 = \langle 4, (28, 12), (198, 40) \rangle$and$\mathbf{CF}_3 = \langle 1, (7, 9), (49, 81) \rangle.\n * Root Node summarizes the children with aggregated vectors: \mathbf{CF}{12} = \langle 9, (44, 42), (252, 230) \rangle$and$\mathbf{CF}3 = \langle 1, (7, 9), (49, 81) \rangle.\n\n* **Practice Exercise Specifications:**\n * *Sequence 1:* Arriving 2D points P_1 = (1, 1), P_2 = (2, 2), P_3 = (5, 5), P_4 = (6, 5)$with radius threshold$T \le 1.0$and branching factor$B = 2.\n * *Sequence 2:* Arriving 2D points P_1 = (1, 1), P_2 = (2, 2), P_3 = (5, 5), P_4 = (6, 5), P_5 = (6, 6), P_6 = (10, 10), P_7 = (11, 10)$with$T \le 1.0$and$B = 2$. Points cluster into distinct groups$(P_1, P_2)$,$(P_3, P_4, P_5)$, and$(P_6, P_7), triggering sequential leaf and non-leaf node splits across the hierarchical CF-Tree.\n\n# Stream Join Processing and Window Joins\n\n* **Fundamentals of Stream Joins:**\n * Streaming data sources are unbounded, preventing traditional relational join operators from blocking until all input is consumed.\n * Join processing must operate continuously, matching arriving tuples against active temporal windows while promptly purging expired records.\n\n* **Taxonomy of Stream Join Implementations:**\n * **Window Join:** Evaluates incoming tuples from two or more streams across a fixed time window constraint (e.g., joining temperature readings from Stream A with location readings from Stream B arriving within the same 5-second interval).\n * **Symmetric Hash Join:** Concurrently maintains two separate in-memory hash tables (one for each stream). As a new tuple arrives on either stream, it inserts itself into its respective hash table and simultaneously probes the opposite stream's hash table for matches.\n * **Sliding Window Join:** Continuously moves temporal boundaries forward as time advances, dynamically updating the candidate set for correlation.\n * **Stateful Join:** Retains intermediate operator state over extended durations to join events that experience significant out-of-order delivery or operational latency.\n * **Supported Systems:** Modern stream processing systems that implement stream joins and indexing include Apache Flink, Apache Kafka Streams, Apache Storm, Google Dataflow, and Spark Structured Streaming.\n\n* **The COMPUTEJOIN Algorithm:**\n * An asymmetric, multi-way streaming window join algorithm executed whenever a new tuple k$arrives from window$S_i.\n\n\n\n * *Algorithmic Logic:*\n * Input: A newly arrived tuple k$from window$S_i$, evaluated against a fixed join order$(S_1, \dots, S{i-1}, S_{i+1}, \dots, S_n).\n * Step 1: Probe window S_1$. For every tuple$u \in S_1, verify that its timestamp falls within the temporal boundary: \n      k.ts - T_1 \le u.ts \le k.ts\n      and verify the join attribute predicate: k.attr \;\theta\; u.attr.\n * Step 2 to i-1$: Loop sequentially through preceding windows$S_2$through$S_{i-1}$. For every tuple$v \in S_{i-1}, verify: \n      k.ts - T_{i-1} \le v.ts \le k.ts \quad \wedge \quad k.attr \;\theta\; v.attr\n * Step i+1$to$n$: Loop sequentially through succeeding windows$S_{i+1}$through$S_n$. For every tuple$w \in S_{i+1}, verify: \n      k.ts - T_{i+1} \le w.ts \le k.ts \quad \wedge \quad k.attr \;\theta\; w.attr\n * Step n$: For every matching tuple$x \in S_n$satisfying$k.ts - T_n \le x.ts \le k.ts$and$k.attr \;\theta\; x.attr, return the combined concatenated tuple: \n      k \circ u \circ v \circ \dots \circ x\n\n* **Two-Stream Sensor Join Example:**\n * *Stream S_1 (Soil Moisture):*\n * Tuple 1: `(A, 30%, ts = 8)`\n * Tuple 2: `(A, 40%, ts = 12)`\n * *Stream S_2 (Temperature):*\n * New arriving tuple: k = (A, 35^\circ\text{C}, ts = 15)\n * *Join Constraints:* Join on `Zone` where timestamps satisfy a 5-second window (T = 5):\n    k.ts - 5 \le u.ts \le k.ts \implies 15 - 5 \le u.ts \le 15 \implies 10 \le u.ts \le 15\n * *Evaluation against S_1:*\n * Tuple `(A, 30%, 8)`: Timestamp 8 otin [10, 15]$$\rightarrow Rejected.\n * Tuple `(A, 40%, 12)`: Timestamp 12 \in [10, 15]$and Zone matches `A`$\rightarrow Accepted.\n * *Emitted Output:* `Zone: A, Moisture: 40%, Temperature: 35°C`.\n\n* **Three-Stream Multi-Way Window Join Example:**\n * Given three streaming windows with temporal limits T_1 = T_2 = T_3 = 10\,\text{seconds}, joined on `StudentID`:\n * Window S_1 (Student Records):\n * `(StudentID: 101, Student Name: Arun, Timestamp: 92)`\n * `(StudentID: 102, Student Name: Anu, Timestamp: 95)`\n * `(StudentID: 103, Student Name: Rahul, Timestamp: 98)`\n * Window S_2 (Exam Records):\n * `(StudentID: 101, Subject: Mathematics, Timestamp: 100)`\n * `(StudentID: 102, Subject: Science, Timestamp: 103)`\n * Window S_3 (Course Records):\n * `(StudentID: 101, Course: Computer Science, Timestamp: 94)`\n * `(StudentID: 102, Course: Commerce, Timestamp: 97)`\n * `(StudentID: 101, Course: Data Science, Timestamp: 89)`\n * *Execution Trace:* A new tuple arrives in S_2: \n    k = (101, \text{Mathematics}, k.ts = 100)\n * Calculate eligible timestamp window for all candidate records: \n      k.ts - T_i \le u.ts \le k.ts \implies 100 - 10 \le u.ts \le 100 \implies 90 \le u.ts \le 100\n * **Search Window S_1:**\n * Tuple `(101, Arun, 92)`: 92 \in [90, 100]$and ID = 101$\rightarrow$Accepted ($u = (101, \text{Arun}, 92)).\n * Tuple `(102, Anu, 95)`: ID = 102 \rightarrow Rejected.\n * Tuple `(103, Rahul, 98)`: ID = 103 \rightarrow Rejected.\n * **Search Window S_3:**\n * Tuple `(101, Computer Science, 94)`: 94 \in [90, 100]$and ID = 101$\rightarrow$Accepted ($w = (101, \text{Computer Science}, 94)).\n * Tuple `(102, Commerce, 97)`: ID = 102 \rightarrow Rejected.\n * Tuple `(101, Data Science, 89)`: Timestamp 89 < 90$(outside window)$\rightarrow Rejected.\n * **Final Joined Result:** Concatenates matching records k \circ u \circ w:\n * `StudentID: 101, Student Name: Arun, Subject: Mathematics, Course: Computer Science`.\n\n# Principles of Complex Event Processing (CEP)\n\n* **Foundational Concepts:**\n * Complex Event Processing (CEP) comprises technologies for processing, filtering, and correlating heterogeneous, distributed data streams in real time to detect high-level, situational situations as they occur.\n * **Event:** A timestamped observation or record of an activity or state change in an external system (e.g., a financial transaction, sensor measurement, button click, or system log).\n * **Simple (Primitive) Event:** An atomic, indivisible occurrence captured directly from a source at a single instant: \n    `"Sensor S3 read 92°C at 10:02:14"`, `"Card swiped at Terminal 7"`, `"HTTP 500 error logged"`.\n * **Complex (Derived) Event:** A higher-level abstraction synthesized by combining, correlating, or aggregating multiple primitive or complex events over a temporal boundary: \n    `"3 failed logins in 60s → possible intrusion"`, `"Rising temp + smoke → fire risk"`, `"Price drop + volume spike → buy signal"`.\n\n* **Architectural Paradigm Shift (DBMS vs. CEP):**\n * Traditional database management systems operate on a *store-then-query* model: data is written to disk/memory tables, and passive queries are executed on demand against static, stored datasets.\n * CEP operates on a *continuous dataflow* model: long-standing queries are deployed, and transient, streaming data in motion is executed continuously through them.\n\n\n\n* **Core Motivations for CEP Adoption:**\n * **Ultra-Low Latency:** Millisecond or sub-millisecond response times required to take automated action before state changes become irreversible.\n * **High Input Volume:** Capability to ingest, filter, and process hundreds to thousands of events per second per node.\n * **Complex Temporal and Spatial Relationships:** Capability to detect multi-event sequences across distributed networks with spatial or temporal dependencies.\n\n* **Data Stream Processing vs. Complex Event Processing Models:**\n * *Data Stream Processing (DSP):* Originating from database extensions, DSP focuses on transforming and aggregating streaming tuples to produce new derived streams (e.g., rolling averages, stream projections).\n * *Complex Event Processing (CEP):* Focuses on treating incoming items as discrete notifications of real-world occurrences that must be matched against declarative patterns and causal chains to infer emergent phenomena.\n\n\n\n* **Real-World Event Correlation Scenarios:**\n * *Building Fire Detection:* A fire alert requires three independent sensor observations located within an area smaller than 100\,\text{m}^2$reporting temperatures greater than$60^\circ\text{C}$within$10\,\text{seconds} of each other. Raw sensor readings carrying normal values are discarded immediately; relevant readings are retained only long enough to confirm pattern matches, and discarded once the alert is generated.\n * *Intrusion Detection Pattern:* \n    `DoorOpened(MainDoor)` followed by `MotionDetected(LivingRoom)`, with no `AuthorizedPersonDetected` within 30\,\text{seconds}.\n * *Glacial Lake Outburst Flood (GLOF) Case Study (Langtang Valley, Nepal-China Border, August 26, 2026):*\n * Event Sequence:\n * March 2026: Satellite detection of a newly formed supraglacial lake on top of a glacier.\n * August 26, 2026, 8:37 AM: A magnitude 5.2 seismic event is recorded across seismometer arrays.\n * August 26, 2026, 8:40 AM: A catastrophic ice-rock avalanche breaks away high above the border, plunging 1,200 meters into the valley below.\n * Within minutes, the debris flow dams the Lhende Khola river.\n * Within an hour, the natural dam breaches, releasing a devastating debris flood across Timure, Syapru Besi, and down the Trishuli River towards India.\n * 8:42 AM: Downstream automated weather stations (AWS) cease responding.\n * Why Conventional Warnings Failed: Automated early warning systems relied exclusively on precipitation gauges. Because there was zero rainfall, rainfall-triggered flood alarms never sounded.\n * Formulated CEP Rule: \n ```sql\n PATTERN glof_risk =\n (GlacierDisplacement.rate > 5 * baseline)\n -> NOT (Rainfall.mm > 5)\n WITHIN 30 days\n GROUP BY basin_id\n      ```\n\n\n\n* *Monsoon Flash Flood and Landslide Case Study (Kerala, July 31 – August 1, 2026):*\n * Heavy precipitation begins around 11:00 PM on Friday, July 31, 2026. Within a 24-hour window ending Saturday, August 1, 2026, Vagamon records 314\,\text{mm}$of rain and Pala records$179.4\,\text{mm}.\n * The Indian Meteorological Department (IMD) red alert threshold is set at rainfall exceeding 204.4\,\text{mm} in 24 hours.\n * The extreme downpour triggers simultaneous flash floods and landslides, sweeping vehicles into mudslides on the Eeratupetta-Vagamon road and raising the Meenachil River past warning levels.\n * Formulated CEP Rule: \n ```sql\n PATTERN flash_flood =\n (Rainfall.mm24h > 200) AND (RiverLevel > danger_mark) AND (LandslideReport = true)\n WITHIN 12 hours\n GROUP BY taluk\n      ```\n\n# CEP Specification Languages, Operators, and Formal Semantics\n\n* **The Four Fundamental CEP Models:**\n\n\n\n 1. **Data Model:** Represents the stream S$as an unbounded, continuous sequence of structured tuples$e$. Each event possesses an explicit timestamp attribute$e.ts$(monotonically increasing) and an event type$T_i$belonging to the universe$\epsilon = {T_1, \dots, T_n}.\n 2. **Language Model:** Provides declarative pattern specification languages (PSLs) typically adopting SQL-like syntax (e.g., Esper EPL, SiddhiQL) or programmatic APIs to specify pattern matching logic.\n 3. **Processing Model:** Governs the internal stateful automata, transition trees, or dataflow networks executed across Event Processing Agents (EPAs).\n 4. **Time Model:** Distinguishes between **Event Time** (the exact timestamp at which the event physically occurred at its source) and **Processing Time** (the system clock timestamp when the processing node observes the event).\n\n* **Formal Definitions of Core CEP Operators:**\n * **Filtering (Unary Predicate):** Evaluates a predicate over individual incoming events, dropping non-matching items immediately without state retention: \n    \text{FILTER}{\phi}(S) = {e \in S \mid \phi(e) = \text{true}}\n * **Sequence (Binary Temporal Operator):** Matches when event e_1 \in T_1$is followed strictly in time by event$e_2 \in T_2$within temporal window$W = [ts_b, ts_e)$. Specified with the keyword `SEQ` or semicolon `;`. Sequence is **associative** but **non-commutative** ($T_1 ; T_2 eq T_2 ; T_1): \n    (T_1 ; T_2)^* = {(e_1, e_2) \mid e_1 \in T_1 \;\wedge\; e_2 \in T_2 \;\wedge\; e_1.ts < e_2.ts \;\wedge\; e_1.ts, e_2.ts \in [ts_b, ts_e)}\n * **Conjunction (Binary Logical Operator):** Requires the occurrence of both events e_1 \in T_1$and$e_2 \in T_2$within window$W, regardless of their relative arrival order. Conjunction is both **associative** and **commutative**: \n    [(T_1 \wedge T_2)]{ts_b}^{ts_e} := \exists ts_n, ts_m : T_1(ts_n) \;\wedge\; T_2(ts_m) \;\wedge\; ts_n, ts_m \in [ts_b, ts_e)\n * **Disjunction (Binary Alternative Operator):** Matches if either event e_1 \in T_1$or event$e_2 \in T_2$occurs within window$W. Specified with the keyword `OR`. Disjunction is **associative** and **commutative**.\n * **Iteration (Kleene Closure Operator):** Matches a repeating sequence of m$occurrences ($m > 0$) of events of type$T under monotonically increasing timestamps: \n    (T^m)^* = {(e_1, \dots, e_m) \mid \forall 1 \le i \le m : e_i \in T \;\wedge\; (e_1.ts < \dots < e_m.ts)}\n * **Negation (Unary Absence Operator):** Specifies the non-occurrence of any event of type T$throughout the duration of window$W.\n\n* **Concrete Operator Execution Traces:**\n * *Filtering:* Smart thermostat stream with overheat condition (`value >= 75°F`). Stream: `[68, 72, 80, 65, 90, 74]`. Entries at index 2 (`80°F`) and index 4 (`90°F`) are forwarded; all others are dropped immediately.\n * *Sequence (A \rightarrow B in order):* Activity stream detecting login (`L`) followed later by password reset (`P`). Stream: `[L, X, P, X, L, P]`. Matches: `(L@0 -> P@2)` and `(L@4 -> P@5)`. Any `P` occurring prior to an `L` is ignored.\n * *Iteration (A^+ Closure):* E-commerce checkout flow grouping repeated additions to cart (`A`) culminating in checkout (`C`). Stream: `[A, A, A, C, A, C]`. Yields match group 1: `(A, A, A) -> C` (3 items) and match group 2: `(A) -> C` (1 item).\n * *Conjunction (A \wedge B, any order):* Home security alarm firing when both Motion (`M`) and Door Open (`D`) occur. Stream: `[X, M, X, D, D, M]`. Match 1 fires at index 3 (`M@1` already seen, now `D@3` arrives); Match 2 fires at index 5 (`D@4` seen, now `M@5` arrives).\n * *Negation (A$NOT followed by$B):* Payment monitoring where a payment event `P` without a confirmation event `C` within 3 steps triggers a timeout alert. Stream: `[P, C, P, X, X, X]`. `P@0` is confirmed by `C@1`. `P@2` reaches index 5 without confirmation, triggering an alert.\n * *Join (Two Streams by Key):* Orders stream joined with Payments stream on `order_id`:\n * Orders: `(id 7, $250)`, `(id 3, $120)`, `(id 9, 75)`.\n * Payments: `(id 3, Paid)`, `(id 9, Paid)`, `(id 7, Pending)`.\n * Joined Result: `{7, 250, Pending}`, `{3, $120, Paid}`, `{9, $75, Paid}`.\n * *ATM Fraud Multi-Operator Pattern:* Detects suspected ATM fraud through a composition of operators:\n * Predicate Filtering: First transaction `amount > ₹50,000`.\n * Sequence: High-value transaction followed sequentially by repeated failed transactions, terminating in account lock.\n * Iteration: Multiple failed authentication attempts (\text{Failure}^+$).
  • Correlation: All events linked by identical account_number.
  • Disjunction: Pattern triggered by an ATM transaction OR an Online transaction.

• Apache Flink CEP Implementation Example:

  • The following program defines an execution environment, ingests a sensor stream, specifies a temperature spike pattern of three consecutive readings above $28.0^\circ\text{C}$ within a 10-second window, and prints alerts:

public class CEPExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Ingest streaming temperature events
        DataStream<TemperatureEvent> temperatureStream = env.fromElements(
            new TemperatureEvent(1L, 20.0),
            new TemperatureEvent(2L, 21.0),
            new TemperatureEvent(3L, 22.0),
            new TemperatureEvent(4L, 30.0),
            new TemperatureEvent(5L, 25.0),
            new TemperatureEvent(6L, 22.0)
        );

        // Formulate pattern: 3 consecutive readings > 28.0C within 10 seconds
        Pattern<TemperatureEvent, ?> temperatureSpikePattern = Pattern.<TemperatureEvent>begin("start")
            .where(event -> event.getTemperature() > 28.0).times(3)
            .within(Time.seconds(10));

        // Apply pattern to stream and generate notification alert
        DataStream<String> spikeAlerts = CEP.pattern(temperatureStream, temperatureSpikePattern)
            .select(pattern -> "Temperature spike detected! Three consecutive readings above 28 degrees.");

        spikeAlerts.print();
        env.execute("CEP Example");
    }
}

• CEP Architectural Guidelines:
  • System designs must explicitly evaluate: Centralized vs. distributed broker topologies, external data integration, event ingestion volume, out-of-order event uncertainty and network ambiguity, processing latency performance, end-to-end security/encryption, and continuous monitoring and debugging.

Windowing Mechanisms in Stream Processing

• Concept of Windowing:
  • Windowing divides an unbounded (or bounded) stream into finite temporal chunks over which stateful computations, aggregations, and pattern checks can be applied.

• Tumbling (Fixed) Windows:

  • Segments stream history into contiguous, non-overlapping intervals of fixed temporal duration applied uniformly across all keys (aligned windows).
  • Every event belongs to exactly one tumbling window:
    $\text{Window Index} = \left\lfloor \frac{t}{\text{Window Size}} \right\rfloor$
  • Query Example (Azure Stream Analytics):sql SELECT TimeZone, COUNT(*) AS Count FROM TwitterStream TIMESTAMP BY CreatedAt GROUP BY TimeZone, TumblingWindow(second, 10)     
  • Worked Trace: Given readings at timestamps $t = 3, 7, 12, 14, 19, 23, 26$ across a 10-second tumbling window:
  • Window $[0, 10)$: Contains $t = 3, 7 \implies \text{Count} = 2$.
  • Window $[10, 20)$: Contains $t = 12, 14, 19 \implies \text{Count} = 3$.
  • Window $[20, 30)$: Contains $t = 23, 26 \implies \text{Count} = 2$.

• Hopping Windows:

  • Slices time into fixed-length windows that advance forward by a fixed hop period. When the hop advance period is smaller than the window length, consecutive windows overlap.
  • Each arriving event falls into multiple overlapping windows:
    $\text{Number of Windows per Event} = \frac{\text{Window Size}}{\text{Hop Size}}$
  • If hop size equals window size, the hopping window degenerates into a tumbling window.
  • Worked Trace: Window length 10 s, hop advance 5 s starting at $t = 0$. Orders arrive at $t = 6, 8, 12, 17$. Each order belongs to $\frac{10}{5} = 2$ windows:
  • Window $[0, 10)$: Contains $t = 6, 8 \implies \text{Count} = 2$.
  • Window $[5, 15)$: Contains $t = 6, 8, 12 \implies \text{Count} = 3$.
  • Window $[10, 20)$: Contains $t = 12, 17 \implies \text{Count} = 2$.
  • Window $[15, 25)$: Contains $t = 17 \implies \text{Count} = 1$.

• Sliding Windows:

  • Windows that evaluate continuous boundaries, emitting output only at discrete instants when the internal contents of the window change (i.e., when an event enters or expires).
  • Query Example: Alert when a topic is mentioned at least 3 times within a 10-second interval:
    sql SELECT Topic, COUNT(*) FROM TwitterStream TIMESTAMP BY CreatedAt GROUP BY Topic, SlidingWindow(second, 10) HAVING COUNT(*) >= 3     
  • Worked Trace: Failed logins arrive at $t = 6, 8, 12, 27$. Window length 10 s; alert threshold $\text{Count} \ge 3$.
  • Event lifespans: Event 6 exits at 16; Event 8 exits at 18; Event 12 exits at 22; Event 27 exits at 37.
  • Output evaluation points:
    
    • At $t = 6$: Count = 1
    • At $t = 8$: Count = 2
    • At $t = 12$: Count = 3 $\rightarrow$ Alert fires at $t = 12$
    • At $t = 16$: Count drops to 2
    • At $t = 18$: Count drops to 1
    • At $t = 22$: Count drops to 0 (empty, no emission)
    • At $t = 27$: Count = 1
    • At $t = 37$: Count drops to 0
  • In contrast, a 10-second tumbling window would observe counts of 2 in $[0, 10)$ and 1 in $[10, 20)$, failing to detect the breach.

• Session Windows:

  • Dynamic, unaligned windows that demarcate periods of user or device activity bounded by gaps of inactivity greater than a specified timeout.
  • Controlled by three parameters: Inactivity Timeout, Maximum Duration, and optional Partitioning Key.
  • Operational Mechanics: A session window opens on the first event. If a subsequent event arrives within the timeout interval, the window extends. If no event arrives before the timeout expires, the window closes. If events arrive continuously, the window closes once the maximum duration boundary is reached.

• Query Example:sql SELECT System.Timestamp() AS WindowEnd, Topic, COUNT(*) FROM TwitterStream TIMESTAMP BY CreatedAt GROUP BY Topic, SessionWindow(second, 5, 10)     

  • Worked Trace: Clicks occur at minutes 0, 2, 4, 12, 13, 25 with a 5-minute timeout:
  • Session 1 ($S_1$): Clicks at 0, 2, 4. Gap between 4 and 12 is 8 min ($> 5$ min). Closes with 3 events; end time = 9.
  • Session 2 ($S_2$): Clicks at 12, 13. Gap between 13 and 25 is 12 min ($> 5$ min). Closes with 2 events; end time = 18.
  • Session 3 ($S_3$): Click at 25. Closes with 1 event; end time = 30.

• Snapshot Windows:

  • Open and close instantaneously to group events sharing the exact same physical timestamp.

• Query Example:sql SELECT System.Timestamp() AS WindowEnd, Topic, COUNT(*) FROM TwitterStream TIMESTAMP BY CreatedAt GROUP BY Topic, System.Timestamp()     

  • Worked Trace: Device readings (time, device): (1, A), (1, B), (2, A), (4, A), (4, B), (4, C), (7, B).
  • Grouping by identical timestamp yields 4 snapshots:
    
    • At $t = 1$: Devices A, B \implies \text{Count} = 2$\n * At t = 2$: Device A$\implies \text{Count} = 1$
    • At $t = 4$: Devices A, B, C \implies \text{Count} = 3$\n * At t = 7$: Device B$\implies \text{Count} = 1$
  • No snapshots are emitted at unpopulated timestamps ($t = 3, 5, 6$).

• Application Windowing Matrix:

  • Daily fitness step tracker reset at midnight $\rightarrow$ Tumbling Window.
  • Security freeze after 3 incorrect PIN entries within 5 minutes $\rightarrow$ Sliding Window.
  • Media viewing sessions ending after an idle interval $\rightarrow$ Session Window.
  • Weather forecast displaying 30-minute rain probability updated every 5 minutes $\rightarrow$ Hopping Window.
  • Live sports leaderboard displaying player statistics at the instant of page refresh $\rightarrow$ Snapshot Window.

Apache Kafka Architecture and Distributed Event Streaming Ecosystem

• System Overview:
  • Originally developed by LinkedIn and managed by the Apache Software Foundation, Apache Kafka is an open-source, horizontally scalable, fault-tolerant distributed event streaming platform designed to publish, subscribe to, store, and process streams of records in real time.

• Core Architectural Components:

  • Producers: Client applications that publish (push) structured event messages into Kafka topics.
  • Consumers: Client applications that pull (poll) messages from topics, processing records asynchronously.
  • Brokers: Stateless or state-managed server nodes forming the Kafka storage cluster. Brokers manage append-only commit logs, service write and read requests, and handle data replication.
  • Topics: Logical categories or feed names to which records are published. Topics are subdivided into partitions.
  • Partitions: The fundamental unit of parallelism and scale in Kafka. Each partition is an ordered, immutable sequence of messages continuously appended to a structured commit log.
  • Consumer Groups: Sets of consumers cooperating to consume data from a topic. Each partition within a topic is consumed by exactly one consumer within a given consumer group, ensuring ordered processing per partition while scaling read throughput horizontally.
  • Coordination Engine: Historically managed by Apache ZooKeeper for cluster metadata, broker health tracking, and partition leader elections. Modern deployments utilize KRaft (Kafka Raft Metadata Mode), an integrated Raft-based consensus protocol eliminating the external ZooKeeper dependency.

• Log Storage Internals (Segments and Indexes):

  • Partitions are stored physically on broker disks as a series of segment files paired with sparse index files.
  • Arriving messages are appended to the active segment log with an incrementing 64-bit integer called an offset.

• Index files map logical offsets to raw physical byte positions within the segment file, enabling random access in $O(1)$ time via memory-mapped files (mmap).

• Offset Management: Consumers track progress by updating their committed offset within an internal topic (__consumer_offsets). When a consumer reads offset 3, its next poll operation requests offset 4.

  • Partitioning Guarantees and Ordering Semantics:

• Total message ordering is guaranteed strictly within an individual partition, not globally across an entire topic.

• By setting an explicit message key, Kafka's hashing partitioner guarantees that all events sharing that key map to the exact same partition, preserving execution and causality order.

• Python Producer Implementation:

  from kafka import KafkaProducer
  producer = KafkaProducer(bootstrap_servers='localhost:9092')
  
  # Messages sharing key user_123 route to the same partition and preserve strict order
  producer.send('user-events', key=b'user_123', value=b'{"action": "login"}')
  producer.send('user-events', key=b'user_123', value=b'{"action": "purchase"}')
      ```
  * *Python Consumer Implementation:*  
  

  python from kafka import KafkaConsumer consumer = KafkaConsumer( 'user-events', bootstrap_servers='localhost:9092', group_id='analytics-service', auto_offset_reset='earliest' ) for message in consumer: print(f"Partition: {message.partition}, Offset: {message.offset}, Value: {message.value}") process_user_event(message.value)     ```

  • Replication and Fault Tolerance:

• Each partition is replicated across multiple brokers according to a configured replication factor.

• For each partition, one broker is elected as the Partition Leader, while remaining assigned brokers act as Follower Replicas (Copies).

• All producer writes and consumer reads are handled by the partition leader. Followers synchronize log segments by pulling data from the leader to maintain In-Sync Replica (ISR) status. If the leader fails, a follower in the ISR set is promoted to leader.

  • Extended Ecosystem Components:

• Kafka Connect: A scalable framework for streaming data between Kafka and external datastores via pre-built connectors. Source Connectors ingest from databases (via JDBC), Couchbase, Cassandra, or AWS S3 into Kafka; Sink Connectors export Kafka topics into downstream analytics engines, Elasticsearch, or data warehouses.

• Kafka Streams: A client library for building real-time stream processing applications directly on top of Kafka. Supports stateful stream transformations, windowing, stream-table duality ($KStream$ and $KTable$), stream-to-stream joins, and aggregations.

• Kafka Schema Registry: A centralized service that stores Avro, JSON Schema, or Protobuf schemas, ensuring backward and forward compatibility between producers and consumers.

• Kafka REST Proxy: Exposes an HTTP REST interface allowing web applications and non-JVM clients to produce and consume messages using standard JSON payloads.

• MirrorMaker: Replicates data asynchronously across distinct Kafka clusters located in different data centers or cloud regions to establish active-passive or active-active disaster recovery environments (referred to as cluster mirroring to distinguish from internal broker partition replication).