\documentclass[12pt]{article}
%\usepackage{/home/kogan/style/kogan}
\usepackage{kogan}
\usepackage{bm}
%\usepackage{umbc_math}
\usepackage{epsfig}
\setlength{\oddsidemargin}{-0.55cm}
\setlength{\topmargin}{-1.05in}
\setlength{\textwidth}{17.3cm}
\setlength{\textheight}{9.9in}
\begin{document}
\baselineskip 20 pt
\thispagestyle{empty}
%\input{head}
\begin{center}
October 12, 2013 -DK1JK1-
%\today -1-
\end{center}
\begin{center}
{\Large
Monitoring Distributed Data Streams Through Node Clustering}\\
Maria Barouti\footnote{Math. and Stat., UMBC,
Baltimore, MD 21250, bmaria2@umbc.edu},
Daniel Keren\footnote{Department of Computer Science,
Haifa University, Haifa 31905, Israel, 
dkeren@cs.haifa.ac.il},  
Jacob Kogan\footnote{Math. and Stat., UMBC,
Baltimore, MD 21250, kogan@umbc.edu}, 
Yaakov Malinovsky\footnote{Math. and Stat., UMBC, 
Baltimore, MD 21250, yaakovm@umbc.edu}
\end{center}
\begin{abstract}

Monitoring data streams in a distributed system is a challenging 
problem with profound applications. The task of feature selection
(e.g., by monitoring the information gain of various features) 
is an example of an application that requires special techniques 
to avoid a very high communication overhead when performed using 
straightforward centralized algorithms.

Motivated by recent contributions based on geometric ideas, we 
present an alternative approach that combines system theory 
techniques and clustering. The proposed approach enables monitoring 
values of an arbitrary threshold function over distributed data 
streams through a set of constraints applied independently on each 
stream and/or clusters of streams. The clusters are designed to adopt 
themselves to the data stream. A correct choice of clusters yields 
a reduction in communication load. Unlike many clustering algorithms 
that attempt to collect together similar data items monitoring requires
clusters with dissimilar vectors canceling each other as much
as possible. In particular, sub--clusters of a good cluster 
do not have to be good. This novel type of clustering dictated by the 
problem at hand requires development of new algorithms, and the paper 
is a step in this direction.

We report experiments on real-world data that detect instances where 
communication between nodes is required, and show that the clustering 
approach reduces communication load.
\end{abstract}
\bigskip

\noindent
{\bf Keyword list}: data streams, convex analysis,
distributed system, clustering
%\input{intro}
\section{Introduction}
\label{intro}
In many emerging applications one needs to process a continuous stream 
of data in real time. Sensor networks
\cite{madden:franklin:2002},
network monitoring
\cite{dilman:raz:2001},
%intrusion detection, 
and
real--time analysis of financial data
\cite{yi:sidiropoulos:johnson:jagadish:faloutsos:biliris:2000},
\cite{zhu:shasha:2000}
are examples of such applications. 
Monitoring queries are a particular class of queries in the context of 
data streams. Previous work in this area deals with monitoring simple 
aggregates \cite{dilman:raz:2001}, or term frequency occurrence in a set
of distributed streams \cite{manjhi:shkapenyuk:dhamdhere:olston:2005}.
The current contribution is motivated by results recently reported in 
\cite{Sharfman:Schuster:Keren:2007},
\cite{sharfman:schuster:keren:2010} 
where a more general type of monitoring query is described as follows:

Let $\S=\{\s_1,\dots,\s_n\}$ be a set of data streams collected at $n$ 
nodes $\N=\{\n_1,\dots,\n_n\}$. 
Let $\v_1(t),\dots,\v_n(t)$ be  $d$-dimensional, real-valued,
time varying vectors derived from the streams. For a function 
$f\;:\;\R^d\to \R$ 
we would like to monitor the inequality 
\begin{equation}
\label{intro.eq.boundary}
f\left({{\v_1(t)+\dots +\v_n(t)}\over n}\right)>0
\end{equation}
while minimizing communication between the nodes. 
Often the threshold might be a constant $r$ other than $0$. 
In what follows, 
for notational convenience, 
we shall always consider the inequality $f>0$,
and when one is interested in monitoring the inequality $f>r$ we will
modify the threshold function and consider $g=f-r$, so that the inequality
$g>0$ yields $f>r$.% (see $(\ref{ig.eq.fr})$).  ~
~In e.g. \cite{Sharfman:Schuster:Keren:2007,DBLP:conf/vldb/GabelKS13,sketchvldb13,DBLP:conf/icde/BurdakisD12}
a few real-life applications of this monitoring problem are described; see also Section
\ref{sec.tma} here.

The difference between monitoring problems involving linear and non-linear 
functions $f$ is discussed 
and illustrated 
by a simple example involving a quadratic function $f$
in \cite{Sharfman:Schuster:Keren:2007}. 
The example demonstrates that, for a non-linear $f$, it is often very difficult 
to determine from the 
values of $f$ at the nodes whether its value evaluated at the average vector is above 
the threshold or not. The present paper deals with the information gain function
(see Section~\ref{sec.tma} for details), and rather 
than focus on the values of $f$ we consider the location of the vectors $\v_i(t)$
relative to the boundary of the the subset of $\R^d$ where $f$ is positive.
We denote this set  by $\Z_+(f)=\{\v\;:\;f(\v)>0\}$, 
and state (\ref{intro.eq.boundary}) as
\begin{equation}
\label{intro.eq.boundary+}
\v(t)={{\v_1(t)+\dots +\v_n(t)}\over n}\in \Z_+(f).
\end{equation}
thus, the functional monitoring problem is transformed to the monitoring of a
{\em geometric} condition.
As a simple illustration, consider the case of three scalar functions 
$v_1(t)$, $v_2(t)$ and $v_3(t)$,
and the identity function $f$ (i.e.
$f(x)=x$). %, and $r=1$. 
We would like to monitor the inequality 
$$
v(t)={{v_1(t)+v_2(t)+v_3(t)}\over 3}>0
$$
while keeping the nodes silent as long as possible.
One strategy is 
to verify the initial inequality 
$\d
v(t_0)={{v_1(t_0)+v_2(t_0)+v_3(t_0)}\over 3}>0
$
and to keep the nodes silent while 
$$
|v_i(t)-v_i(t_0)|<\delta=v(t_0),\;
t\ge t_0,\;
i=1,2,3. 
$$
The first time $t$ when one of the functions, say $v_1(t)$, 
crosses the boundary of the local constraint, i.e. 
$|v_1(t)-v_1(t_0)|\ge \delta$
the nodes communicate, 
$t_1$ is set to be $t$,
the mean $v(t_1)$ is computed, 
the local constraint $\delta$ is updated %as needed
and made available to the nodes. 
The nodes are kept silent as long as the inequalities
$$
|v_i(t)-v_i(t_1)|<\delta,\; t\ge t_1,\; i=1,2,3 
$$
hold. 
This type of monitoring was suggested in \cite{kogan:2012} for a variety 
of vector norms. 
The numerical experiments conducted in \cite{kogan:2012}
with the dataset described in Section~\ref{exp} show that: 
\begin{enumerate}
\item
The number of time instances the mean violates $(\ref{intro.eq.boundary})$
is a small fraction ($<1$\%) of the number of time instances when the
local constraint is violated at the nodes.
\item
The lion's share of communications (about 75\%) is required because of
a single node violation of the local constraint $\delta$.
\item
The smallest number of communications is required when one 
uses the $l_1$ norm.
\end{enumerate}
We note that if, for example, the local constraint is violated at $\n_1$, i.e.
$|v_1(t)-v_1(t_0)|\ge \delta$, and at the same time
$$
v_1(t)-v_1(t_0)=-[v_2(t)-v_2(t_0)],
$$
while $|v_3(t)-v_3(t_0)|< \delta$ 
then 
$|v(t)-v(t_0)|<\delta$, $f(v(t))>0$, and update of the mean can be avoided. 
Separate monitoring 
of the two node cluster $\{\n_1, \n_2\}$ would require communication 
involving 
two nodes only, and could reduce communication load. We aim to extend this
idea to the general case -- involving many nodes, arbitrary functions, and
high-dimensional data.

Clustering in general is a difficult problem, and many clustering problems are
known to be NP-complete \cite{Brucker:1978}. Unlike standard clustering 
that attempts to collect together similar data items
\cite{mirkin:2005}, we are seeking
clusters with {\em dissimilar data items}, which cancel out each other as much as 
possible. While sub-clusters of a ``classical" good cluster are usually good,
this may not be the case when a cluster contains dissimilar objects. These
observations indicate that common clustering methods are not applicable
to our problem.
 
A basic attempt to cluster nodes was suggested in \cite{kogan:malinovsky:2013}
with results reported for the dataset presented in Section~\ref{exp}.
Clustering together just two nodes 
% (the ``longest" and ``shortest" ones) 
reported in \cite{kogan:malinovsky:2013} reduces communication by about 10\%.

In this paper we advance clustering approach to monitoring. 
The main contribution of this work  is twofold: 
\begin{enumerate}
\item
We suggest 
%to cluster nodes in order to reduce communication required, apply 
a specific clustering strategy, and report the communication 
reduction achieved.
\item
We apply the same clustering strategy with $l_1$, $l_2$, and $l_{\infty}$
norms and report the results obtained.
\end{enumerate}
The paper is organized as follows.
In Section~\ref{sec.tma} 
we present a relevant Text Mining application.
Section~\ref{sec.mon.balls} provides motivation for node clustering.
A specific implementation of node clustering is presented in 
Section~\ref{sec.clustering-impl}. 
Experimental results  are reported in Section~\ref{exp}. 
Section~\ref{sec.conc} concludes the paper and indicates new research
directions.
The Appendix details the accounting of message transmission.

In the next section we provide a Text Mining related example that leads 
to a non linear threshold function $f$.

%\input{tma}

\section{Text Mining application}
\label{sec.tma}
Let $\T$ be a  textual database (for example a collection of mail 
or news items). 
We denote the size of the set $\T$ by $|\T|$. 
We will be concerned with two subsets of $\T$:
\begin{enumerate}
\item
$\R$--the set of ``relevant" texts (e.g. texts not labeled as "spam"),
\item
$\F$--the set of texts that contain a ``feature" (word or term for example).
\end{enumerate}
We denote complements of the sets by $\overline{\R}$, $\overline{\F}$ 
respectively (i.e. $\R\cup \overline{\R}=\F\cup \overline{\F}=\T$), 
and consider the relative size of the four sets 
$\F\cap \overline{\R}$,
${\F}\cap {\R}$,
$\overline{\F}\cap \overline{\R}$,
and
$\overline{\F}\cap {\R}$
as follows:
\begin{equation}
\label{ig.4sers}
\begin{array}{ll}
%x_{11}=
x_{11}(\T)={\d {{|\F\cap \overline{\R}|}\over {|\T|}}},
&
%x_{12}=
x_{12}(\T)={\d {{|\F\cap \R|}\over {|\T|}}},
\\
 & \\
%x_{21}=
x_{21}(\T)={\d {{|\overline{\F}\cap \overline{\R}|}\over {|\T|}}},
&
%x_{22}=
x_{22}(\T)={\d {{|\overline{\F}\cap \R|}\over {|\T|}}}.
\end{array}
\end{equation}
Note that 
$$
0\le x_{ij} \le 1,
\hbox{ and }
x_{11}+x_{12}+x_{21}+x_{22}=1.
$$
The function 
%$f\;:\;\R^4\to \R^1$
$f$
is defined on the simplex (i.e. $x_{ij}\ge 0$, $\d \sum x_{ij}=1$),
and given by
\begin{equation}
\label{ig.ig-function}
%f(x_{11},x_{12},x_{21},x_{22}) =
\sum_{i,j}x_{ij}\log
\left(
{x_{ij}}
\over
{(x_{i1}+x_{i2})(x_{1j}+x_{2j})}
\right),
\end{equation}
where $\log x=\log_2 x$ throughout the paper.
It is well-known that (\ref{ig.ig-function})
provides the {\em information gain} for the ``feature"
(see e.g. \cite{gray:1990}).

As an example, we consider $n$ agents installed on $n$ 
different servers, and a stream of texts arriving at the servers. 
Let $\T_h=\{\t_{h1},\dots,\t_{hw}\}$ be the last
$w$ texts received at the $h^{th}$ server, with $\d \T=\bigcup_{h=1}^n \T_h$.
Note that 
$$
x_{ij}(\T)=
\sum_{h=1}^n
{{|\T_h|}\over {|\T|}}
x_{ij}(\T_h),
$$
i.e., entries of the global contingency table $\{x_{ij}(\T)\}$ are the 
weighted average
of the local contingency tables $\{x_{ij}(\T_h)\}$, $h=1,\dots,n$.

To check that the given ``feature" is sufficiently informative with 
respect to the target relevance label $r$, one may want to monitor 
%For the given ``feature" and a predefined positive threshold $r$ we would 
%like to verify 
the inequality
\begin{equation}
\label{ig.eq.fr}
f\left(x_{11}(\T),x_{12}(\T),x_{21}(\T),x_{22}(\T)\right)-r>0
\end{equation}
with $f$ given by $(\ref{ig.ig-function})$
while minimizing communication between the servers.
%Note that $(\ref{ig.ig-function})$ is a nonlinear function.
%The case of a nonlinear monitoring function is %fundamentally 
%different from that of linear one %\cite{Sharfman:Schuster:Keren:2007}
%(in fact \cite{Sharfman:Schuster:Keren:2007}
%calls the nonlinear monitoring function case ``fundamentally different"). 

In the next section we provide motivation to node clustering for monitoring
data streams.
\section{Monitoring threshold functions through clustering: motivation}
\label{sec.mon.balls}
In what follows we denote a norm of a vector $\v$ by $\|\v\|$.
While the experiments reported in this paper have been conducted with
$l_1$, $l_2$, and $l_{\infty}$ norms, the proposed monitoring and node
clustering procedures can be applied with any norm.
%
%\begin{table}
%\begin{center}
%\begin{tabular}{||c|r||}
%\hline
%\hline
%$k$ & $V_k$\\
%\hline
%\hline
%$1$    &3034  \\
%\hline
%$2 $   &620  \\
%\hline
%$3 $   & 162 \\
%\hline
%$4 $   & 70 \\
%\hline
%$5 $   & 38 \\
%\hline
%$6 $   & 26 \\
%\hline
%$7 $   & 34 \\
%\hline
%$8 $   & 17 \\
%\hline
%$9 $   & 5 \\
%\hline
%$10 $  & 0\\ 
%\hline
%\hline
%\end{tabular}
%\caption{number of local constraint violations simultaneously by $k$ 
%nodes and $l_{2}$ norm}
%\label{exp.table.ViolationDistributionA}
%\end{center}
%\end{table}
The monitoring strategy proposed in \cite{kogan:2012} can be briefly 
described as follows:
\begin{algorithm}
\label{sec.mon.balls.al}
Monitoring Threshold Function
\begin{itemize}
\item
A node is designated as a root $\r$.
\item
The root sets $i=0$.
\item
Until end of stream
\begin{enumerate}
\item
The root sends a request to each node $\n$ for the vectors $\v_{\n}(t_i)$.
The nodes respond to the root. The root computes the distance 
$\delta$ between the mean $\d {1\over n}\sum_{\n\in \N}\v_n(t_i)$ and the zero 
set $\Z_f$ of the function $f$. The root transmits $\delta$ to each node.
\item
do for each $\n\in \N$\\
If $||\v_{\n}(t)-\v_{\n}(t_i)||<\delta$\\ 
\hbox{\hskip 1 cm}
the node $\n$ is silent\\
else\\
\hbox{\hskip 1 cm}$\n$ notifies the root about violation of its local constraint
$\delta$\\
\hbox{\hskip 1 cm}the root sets $i=i+1$\\
\hbox{\hskip 1 cm}go to Step 1.
\end{enumerate}
\item
Stop
\end{itemize}
\end{algorithm}
An application of the above procedure to data streams generated from
the Reuters Corpus RCV1--V2 (see Section \ref{exp} for detailed description
of the data and experiments) 
%with $l_2$ norm and the threshold $r=0.0025$ for feature ``bosnia"
leads to $4006$ time instances in which the local constraints are violated, and 
the root is updated. Results presented in 
Table~\ref{exp.table.ViolationDistributionA} 
%yaka
\begin{table}
\begin{center}
\begin{tabular}{||l|c|c|c|c|c|c|c|c|c|c||}
\hline
\hline
\# of nodes violators & 1 &2 &3&4&5&6&7&8&9&10\\
\hline
\# of violation instances & 3034 &620 & 162 & 70 & 38 & 26 & 34 & 17 & 5 & 0\\
\hline
\hline
\end{tabular}
\caption{number of local constraint violations simultaneously by $k$
nodes, $r=0.0025$, $l_{2}$ norm, the feature is ``bosnia"}
\label{exp.table.ViolationDistributionA}
\end{center}
\end{table}
show that in $3034$ out 
of $4006$ time instances, communications with the root are triggered by 
constraint violations at exactly one node.

The results %presented in Table~\ref{exp.table.ViolationDistributionA}
immediately suggest to cluster nodes to further reduce communication
load. %required to monitor the mean. 
Each cluster will be equipped with a ``coordinator" $\c$
(one of the cluster's nodes).
If a cluster node ${\n}$
violates its local constraint at time $t$, 
then the coordinator 
collects vectors $\v_{\n}(t)-\v_{\n}(t_i)$ from all the nodes 
in the cluster, 
computes the mean of the vectors, and checks whether the mean violates the
coordinator constraint $\delta$ (at this point, node and coordinator constraints 
are identical).
We shall follow \cite{Sharfman:Schuster:Keren:2007}
and refer to this step as ``the balancing process."
If the coordinator constraint is violated, the 
coordinator alerts the root, and the mean of the entire dataset is recomputed
by the root (for detailed description of the procedure see 
Section~\ref{sec.clustering-impl}). 

A standard clustering problem is often described as
``\dots finding and describing cohesive or homogeneous chunks in data,
the clusters" (see e.g. \cite{mirkin:2005}). For the problem at hand we 
would like to partition the set of nodes $\N$ into $k$ clusters
$\Pi=\{\pi_1,\dots,\pi_k\}$ so that 
$$
\N=\bigcup_{i=1}^k\pi_i, \hbox{ and }\pi_i\bigcap \pi_j=\emptyset
\hbox{ if }i\not=j.
$$
We denote the size of $\pi_i$ by $|\pi_i|$.
If for each cluster $\pi_i$ one has
$\d {1\over {|\pi_i|}}
\left|\left|
\sum_{\n\in \pi_i} [\v_{\n}(t)-\v_{\n}(t_j)] 
\right|\right|<\delta$,
then, due to convexity of any norm, one has
$$
\left|\left|
{1\over n}\sum_{\n\in\N}\v_{\n}(t)-
{1\over n}\sum_{\n\in\N}\v_{\n}(t_j)
\right|\right|
\le
\sum_{i=1}^k
{{|\pi_i|}\over n}
\left[{1\over {|\pi_i|}}
\left|\left|
\sum_{\n\in \pi_i} [\v_{\n}(t)-\v_{\n}(t_j)] 
\right|\right|
\right]<\delta.
$$
Hence the ``new" mean $\d {1\over n}\sum_{\n\in\N}\v_{\n}(t)$ belongs to 
$\Z_+(f)$ if the ``old" mean $\d {1\over n}\sum_{\n\in\N}\v_{\n}(t_j)$ belongs
to this set.
We therefore may attempt to define the quality of a $k$ cluster
partition $\Pi$ as
\begin{equation}
\label{sec.mon.balls-q}
Q(\Pi)=\max_i
\left\{
{1\over {|\pi_i|}}
\left|\left|
\sum_{\n\in \pi_i} [\v_{\n}(t)-\v_{\n}(t_j)]
\right|\right|\right\},\;i=1,\dots,k.
\end{equation}
Our aim is to identify $k$ and a $k$ cluster
partition $\Pi^o$ that minimizes $(\ref{sec.mon.balls-q})$.
%and to look for $k$ and a $k-$cluster partition $\Pi^o$ that 
%minimizes $Q(\Pi)$.
Our monitoring 
problem requires to assign 
nodes $\{\n_{i_1},\dots,\n_{i_k}\}$ 
to the same cluster $\pi$ 
so that the total average change within cluster 
$$
\d \left|\left|
{1\over {|\pi|}}
\sum_{\n\in \pi} [\v_{\n}(t)-\v_{\n}(t_j)] \right|\right|
\hbox{ for }t>t_j
$$
%for $t>t_j$
is minimized, i.e., nodes with {\bf different} variations
$\v_{\n}(t)-\v_{\n}(t_j)$
that cancel out each other as much as possible are assigned to the
same cluster. Hence, unlike classical clustering procedures, this one needs
to combine ``dissimilar" nodes together. 
% While splitting a ``classical" 
% cluster usually generates two clusters of reasonable quality, this may not
% be the case when a cluster contains dissimilar objects.

The proposed partition quality $Q(\Pi)$ (see $(\ref{sec.mon.balls-q}))$ 
generates three immediate problems:
\begin{enumerate}
\item
Since the arithmetic mean $\overline{a}$ of a finite set of real numbers 
$\{a_1,\dots, a_k\}$ satisfies  
$$
\min \{a_1,\dots, a_k\}\le \overline{a}\le \max\; \{a_1,\dots, a_k\}
$$
the single cluster partition always minimizes $Q(\Pi)$. Considering the entire 
set of nodes as a single cluster with its own coordinator that communicates
with the root introduces an additional unnecessary ``bureaucracy" layer that 
only increases communications. We seek a trade-off which yields clusters
with "good" sizes (this is rigorously defined in Section \ref{sec.clustering-impl}).
\item
Computation of $Q(\Pi)$ involves future values $\v_{\n}(t)$, which are 
not available at time $t_j$ when the clustering is performed.
\item
Since the communication overhead of
the balancing process
%violation recovery stage 
is proportional to the size of a cluster,
the individual clusters' sizes should affect the clustering quality $q(\pi)$.
\end{enumerate}
In the next section we  address these problems.

%\input{mtfballs}


%\input{clustering}

\section{Monitoring threshold functions through clustering: implementation}
\label{sec.clustering-impl}
We argue that in addition to the average magnitude of the variations
$\v_{\n}(t)-\v_{\n}(t_j)$ inside the cluster $\pi$, the cluster's size
also affects the frequency of updates, and, as a result, the communication
load. We therefore define the quality of the cluster $\pi$ by
\begin{equation}
\label{sec.clustering-impl-eq.clus-q}
q(\pi)={1\over {|\pi|}}
\left|\left|
\sum_{\n\in \pi} [\v_{\n}(t)-\v_{\n}(t_j)] 
\right|\right|+\alpha |\pi|,
\end{equation}
where $\alpha$ is a nonnegative scalar parameter. The quality of the partition
$\Pi=\{\pi_1,\dots,\pi_k\}$ is defined by
\begin{equation}
\label{sec.clustering-impl-eq.part-q}
Q(\Pi)=\max_i %\left(
q(\pi_i),%+\alpha |\pi_i|\right).
\end{equation}
When $\alpha =0$ the partition that minimizes $Q(\Pi)$ is a single cluster 
partition (that we would like to avoid). 
When $\d \max_{\n} \left|\left| \v_{\n}(t)-\v_{\n}(t_j)\right|\right|\le \alpha$
the optimal partition is made up of $n$ singleton clusters. In this paper we
focus on
\begin{equation}
\label{sec.clustering-alpha-abs}
0<\alpha <\max_{\n}\left|\left| \v_{\n}(t)-\v_{\n}(t_j)\right|\right|.
\end{equation}
The constant $\alpha$ depends on $t$ and $t_j$, and below we show how to 
avoid %eliminate 
this dependence.
%When $\d \max_i|\pi_i|\le \alpha$
%the optimal partition is made up of $n$ singleton clusters. In this paper we
%focus on $0<\alpha <\max_i|\pi_i|$.

Computation of $Q(\Pi)$ required for the clustering procedure is
described below. In order to compute $Q(\Pi)$ at time $t_j$ one needs to know
$\v_{\n}(t)$ at a future time $t>t_j$ which is not available. While the
future behavior is not known, we shall use past values of $\v_{\n}(t)$ for prediction.
%the past is available. i
For each node $\n$ we build 
``history" vectors $\h_{\n}(t_j)$ defined as follows:
\begin{enumerate}
\item
$\h_{\n}(t_0)=0$
\item
if ($\h_{\n}(t_j)$ is already available)\\
\hbox{\hskip 1cm} $\h_{\n}(t_{j+1})=\h_{\n}(t_j)$\\
\hbox{\hskip 1cm} for $t$ increasing from $t_j$ to $t_{j+1}$ do\\
\hbox{\hskip 2cm} $\d \h_{\n}(t_{j+1})={1\over 2}\h_{\n}(t_{j+1})+
[\v_{\n}(t)-\v_{\n}(t_j)]$
\end{enumerate}
The vectors $\h_{\n}(t_j)$ accumulate the history of changes, with older changes 
assigned smaller weights. We shall use the 
vectors $\{\h_{\n}(t_j)\}$ to generate a
node partition at time $t_j$. We note that normalization of the vector set that
should be clustered does not change the induced optimal partitioning of the 
nodes. When the vector set is normalized by the magnitude of the longest vector
in the set, the range for $\alpha$ conveniently shrinks to $[0,1]$.
In what follows we set
$\d h=\max_{\n}||\h_{\n}(t_j)||$, 
assume that $\d h>0$, 
and describe 
a ``greedy" clustering procedure for the normalized vector set 
$$
\{\a_1,\dots,\a_n\},\;\a_i={1\over h}\h_{\n_i}(t_j),\;i=1,\dots,n.
$$
We start with the $n$ cluster partition $\Pi^{n}$ (each cluster is a 
singleton). If a $k$ cluster partition $\Pi^{k}$, $k>2$ is already available
we 
\begin{enumerate}
\item
identify the partition cluster $\pi_j$ with maximal norm of it's vectors' 
mean, i.e., 
$$
{1\over {|\pi_j|}}\left|\left|\sum_{\a\in \pi_j}\a\right|\right|
\ge
{1\over {|\pi_i|}}\left|\left|\sum_{\a\in \pi_i}\a\right|\right|,
\;i=1,\dots,n.
$$ 
\item
identify cluster $\pi_i$ so that the merger of $\pi_i$ with $\pi_j$
produces a cluster of smallest possible quality, i.e.,
$$
q\left(\pi_j\bigcup \pi_i\right)\le 
q\left(\pi_j\bigcup \pi_l\right),\;l\not=j,
$$
where cluster's quality 
is defined by $(\ref{sec.clustering-impl-eq.clus-q})$.
\end{enumerate}
The partition $\Pi^{k-1}$ is obtained from $\Pi^{k}$ by merging clusters
$\pi_j$ and $\pi_i$. The final partition is selected from the $n-1$ partitions
$\{\Pi^2,\dots,\Pi^n\}$ as the one that minimizes $Q$.

Note that node constraints $\delta$ do not have to be equal.
Taking into account the distribution of the data streams at each node can further
reduce communication.
We illustrate this statement by a simple example involving  two nodes.
If, for example, there is  reason to believe that the inequality
\begin{equation}
\label{exper.eq.weight}
2\|\v_1(t)-\v_1(t_i)\|\le \|\v_2(t)-\v_2(t_i)\|
\end{equation}
always holds,
then the number of node violations may be reduced by imposing node
dependent constraints
$$
\|\v_1(t)-\v_1(t_i)\|<\delta_1={2\over 3}\delta,
\hbox{ and }
\|\v_2(t)-\v_2(t_i)\|<\delta_2={4\over 3}\delta
$$
so that the wider varying stream at the second node enjoys larger
``freedom" of change, while the inequality
$$
\left|\left|
{{\v_1(t)+\v_2(t)}\over 2}
-
{{\v_1(t_i)+\v_2(t_i)}\over 2}
\right|\right|
<{{\delta_1+\delta_2}\over 2}=\delta
$$
holds true. Assigning ``weighted" local constraints requires
information provided by $(\ref{exper.eq.weight})$. With no additional
assumptions about the stream data distribution this information is not
available. Unlike
\cite{keren:sharfman:schuster:livne:2012}
we refrain from making assumptions regarding the underlying data
distributions;
instead, we estimate the weights through past values $\|\v_j(t)-\v_j(t_i)\|$.
\begin{enumerate}
\item
Start with the initial set of weights
\begin{equation}
\label{exper.eq.weight-init}
%w_1(t_0)=\dots =w_n(t_0)=1\;
w_1=\dots =w_n=1\;
\hbox{ and } 
W_1=\dots =W_n=1\;
(\hbox{so that }\sum_{j=1}^nw_j=\sum_{j=1}^nW_j=n).
\end{equation}
\item
As new texts arrive at the next time instance $t$, each node computes
updates
$$
W_j={1\over 2}W_j+\|\v_j(t)-\v_j(t_i)\|, 
\hbox{ with }W_j(t_0)=1,\;j=1,\dots,n.
$$
When at time $t_{i+1}$ the root constraint $\delta(\r)$ needs to be updated,
%a local constraint is violated %, then, in addition to
%$(d+2)|\N|$ messages (see (\ref{ca.eq.mess-brALL})), 
each node $\n_j$ broadcasts
$W_j$ to the root, the root computes
$\d W=\sum_{j=1}^nW_j$, and transmits the updated 
$\delta(\n_j)=w_j\delta(\r)$ where 
$\d
w_j=n\times {{W_j}\over W}\;
(\hbox{so that }\sum_{j=1}^nw_j=n)
$
back to node $j$. 
%so that the node $\n_j$ constraint is $\delta(\n_j)=w_i\delta(\r)$.
For a coordinator $\c$ of a node cluster $\pi$ the %coordinator $\c$ 
constraint $\d \delta(\c)={1\over {|\pi|}}\sum_{\n\in \pi} \delta(\n)$. 
\end{enumerate}
%\input{exp}
\section{Experimental results}
\label{exp}
The data streams analyzed in this section are generated from
the Reuters Corpus RCV1--V2. The data is 
available from 
{\tt http://leon.bottou.org/projects/sgd} and consists of $781,265$ 
tokenized documents with %did (
document ID ranging from 2651 to 810596.
%The methodology described below 
%attempts to follow that presented in 
%\cite{Sharfman:Schuster:Keren:2007}
%and \cite{kogan:2012:1}.
We simulate $n$ streams by arranging the feature vectors in ascending 
order with respect to document ID, %did, 
and selecting feature vectors for the stream
in the round-robin fashion.

In the Reuters Corpus RCV1--V2 each document is labeled as belonging
to one or more categories. We 
%follow \cite{Sharfman:Schuster:Keren:2007} and 
label a vector as ``relevant" if it belongs to 
the 
``CORPORATE/INDUSTRIAL"
(``CCAT")
category, and ``spam" otherwise. 
Following %\cite{sharfman:schuster:keren:2010}
\cite{Sharfman:Schuster:Keren:2007}
we focus on three features:
``bosnia,"
``ipo,"
and
``febru." Each experiment was performed with 10 nodes,
where each node holds a sliding window containing the last 
6,700 documents it received. 

First we use $67,000$ documents to generate initial sliding windows.
The remaining $714,265$ documents are used to generate datastreams, hence
the selected feature information gain is computed $714,265$ times. 
Based on all the documents contained in the sliding window at each one of
the $714,266$ time instances we compute and graph $714,266$ information 
gain values for the feature ``bosnia"
(see Figure~\ref{exp.fig.bosniaNV}).
\begin{figure}[!ht]
\begin{center}
%\epsfig{file=/home/kogan/NUM/2011/keren/bosniaNVline.eps, width=7.0cm}
%\epsfig{file=../keren_pic/bosniaNVline.eps, width=7.0cm}
\epsfig{file=bosniaNVline.eps, width=7.0cm}
%title('nominal signal', 'FontSize', 16)
\end{center}
\caption{information gain values for the feature ``bosnia"}
\label{exp.fig.bosniaNV}
\end{figure}
For the experiments described below, the threshold value $r$
is predefined, and the goal is to monitor the inequality $f(\v)-r>0$
while minimizing communication between the nodes. 
%For convenience
%we denote the set of vectors $\v$ where $f(\v)-r=0$ by $\Z_{f-r}$.

We assume that new texts arrive simultaneously at each node. 
The numerical experiment reported in \cite{kogan:2012}
with the feature ``febru," %the $l_2$ norm,  
and the threshold $r=0.0025$
are shown in Table~\ref{exp.table.febru10sim-1-2-3}
where a broadcast is defined as one time transmission of information
between different nodes.
\begin{table}
\begin{center}
\begin{tabular}{||c|c|r||}
\hline
\hline
norm & mean updates & broadcasts \\
\hline
\hline
$l_1$    & 2591 & 67388 \\
\hline
$l_2$    & 3140 & 81650 \\
\hline
$l_{\infty}$ & 3044 & 79144 \\
\hline
\hline
\end{tabular}
\caption{Number of mean computations, and broadcasts
for feature ``febru" with threshold $r=0.0025$, no clustering
}
\label{exp.table.febru10sim-1-2-3}
\end{center}
\end{table}
We run the node clustering monitoring presented in this paper for the 
same feature and threshold with 
$\alpha=0.05, 0.10,\dots ,0.95$. The best result with respect to 
$\alpha$ is 
presented in Table~\ref{exp.table.febru10sim-m-nc-1-2-3}.
%febru
\begin{table}
\begin{center}
\begin{tabular}{||c|c|c|c|c||}
\hline
\hline
norm & alpha & root mean & coordinator    & total \\
     &       & update    & mean update    & broadcasts\\
\hline
\hline
$l_1$& 0.70  & 1431      & 0              & 38665  \\
\hline
$l_2$& 0.80  & 1317      & 0              & 35597  \\
\hline
$l_{\infty}$
     & 0.65  & 1409      & 0              & 38093  \\
\hline
\hline
\end{tabular}
\caption{Number of root and coordinator mean computations, and
total broadcasts for feature ``febru" with threshold $r=0.0025$
with clustering
}
\label{exp.table.febru10sim-m-nc-1-2-3}
\end{center}
\end{table}
The clustering approach in this case is particularly successful -- coordinators'
constraints are not violated, and the root mean updates are decreased 
significantly. As a result the number of broadcasts decreases by about 
50\%.

Next we turn to the features ``ipo" and ``bosnia." In both cases we run monitoring 
with clustering, allowing $\alpha=0.05, 0.10,\dots ,0.95$, and report results
with the lowest number of broadcasts.
The results obtained 
for ``ipo" without clustering are presented in 
Table~\ref{exp.table.ipo10sim-m}.
%ipo
\begin{table}
\begin{center}
\begin{tabular}{||c|c|r||}
\hline
\hline
norm & mean updates & broadcasts \\
\hline
\hline
$l_1$    & 15331 & 398606 \\
\hline
$l_2$    & 21109 & 548834 \\
\hline
$l_{\infty}$ & 19598 & 509548 \\
\hline
\hline
\end{tabular}
\caption{number of mean computations, and broadcasts
%computed with $l_{2}$ norm
for feature ``ipo" with threshold $r=0.0025$, no clustering
}
\label{exp.table.ipo10sim-m}
\end{center}
\end{table}
Application of clustering procedure leads to  a significant reduction in 
the number of broadcasts. 
Results obtained through clustering procedure are shown in 
Table~\ref{exp.table.ipo10sim-m-nc-1-2-3}.
The table demonstrates significant inside cluster
activity, and a decrease in root mean updates.
%ipo 
\begin{table}
\begin{center}
\begin{tabular}{||c|c|c|c|c||}
\hline
\hline
norm & alpha & root mean & coordinator    & total \\
     &       & update    & mean update    & broadcasts\\
\hline
\hline
$l_1$& 0.15  & 5455      & 829            & 217925  \\
\hline
$l_2$& 0.10  & 7414      & 1782           & 296276  \\
\hline
$l_{\infty}$
     & 0.10  & 9768      & 2346           & 366300  \\
\hline
\hline
\end{tabular}
\caption{Number of root and coordinator mean computations, and
total broadcasts 
for feature ``ipo" with threshold
$r=0.0025$ with clustering 
}
\label{exp.table.ipo10sim-m-nc-1-2-3}
\end{center}
\end{table}

Finally we turn to the feature ``bosnia." Application of clustering to
monitoring this feature information gain appears to be far less successful. 
Results obtained 
without clustering in \cite{kogan:2012} are presented in 
Table~\ref{exp.table.bosnia10sim-m}.
%bosnia
\begin{table}
\begin{center}
\begin{tabular}{||c|c|r||}
\hline
\hline
norm & mean updates & broadcasts \\
\hline
\hline
$l_1$    & 3053 & 79378 \\
\hline
$l_2$    & 4006 & 104156 \\
\hline
$l_{\infty}$ & 3801 & 98826 \\
\hline
\hline
\end{tabular}
\caption{Number of mean computations, and broadcasts,
for feature ``bosnia" with threshold $r=0.0025$, no clustering
}
\label{exp.table.bosnia10sim-m}
\end{center}
\end{table}
Application of the clustering procedure leads to a slight decrease in the 
number of broadcasts
%improvement 
in case of the $l_2$ and $l_{\infty}$ norms (see 
Table~\ref{exp.table.bosnia10sim-m-nc-1-2-3}). In case of the $l_1$ norm,
the 
number of broadcasts increases. Clustering does not offer a universal remedy;
in some cases better performance is achieved with no clustering (by 
keeping $\alpha$ between $0.05$ and $0.95$ we force nodes to 
cluster).
%bosnia
\begin{table}
\begin{center}
\begin{tabular}{||c|c|c|c|c||}
\hline
\hline
norm & alpha & root mean & coordinator   & total \\
     &       & update    & mean update   & broadcasts\\
\hline
\hline
$l_1$& 0.65  & 3290      & 2            & 89128  \\
\hline
$l_2$& 0.55  & 3502      & 7              & 97602 \\
\hline
$l_{\infty}$
     & 0.60  & 3338      & 2            & 91306 \\
\hline
\hline
\end{tabular}
\caption{Number of root and coordinator mean computations, and
total broadcasts 
for feature ``bosnia" with threshold
$r=0.0025$ and clustering
}
\label{exp.table.bosnia10sim-m-nc-1-2-3}
\end{center}
\end{table}
%\input{conc}


\section{Conclusions and future research directions}
\label{sec.conc}
In this paper we propose to monitor threshold functions over distributed 
data streams through clustering nodes whose data fluctuations "cancel out" each other. 
%This approach and by doing so prevent violations of a local constrain 
%by the entire cluster. 
The strategy,
if successful, in many cases reduces the communication required to message 
exchanges within a cluster only, yielding overall communication reduction.

The  clustering strategy suggested is based on minimization
of a combination of average of a vector associated with a cluster and 
the cluster size. The nodes are re--clustered each time the entire 
dataset mean violates its constraint $\delta(\r)$. 
The amount of communication required depends on the 
``trade off" parameter $0<\alpha<1$ selected at the 
beginning of the monitoring process. While the results obtained show 
improvement over previously reported ones that do not use clustering
\cite{kogan:2012}
it is of interest to introduce 
an update of $\alpha$ based on the monitoring history 
each time nodes are re--clustered
(see e.g.  \cite{willems:1971} for feedback theory approach). 

Clustering does not provide a universal remedy. It is of interest to 
identify data streams that benefit from clustering, and those for which
clustering does not reduce communication load in any significant fashion.
Finally a methodology that measures effectiveness of various monitoring 
techniques should be introduced, so that different monitoring strategies can 
be easily compared.
\bibliographystyle{plain}
%\bibliography{/home/kogan/bibfiles/datamine,/home/kogan/bibfiles/keren}
\bibliography{datamine,keren}
%\input{mc.texBR}

\section*{Appendix: counting broadcasts}
\label{mk}
Transmission of a double precision real number is defined as a message in
\cite{kogan:2012}. In this paper, in addition to real numbers typically 
representing vector coordinates, integer values such as node ID and 
node ``reporting order" 
should 
also be transmitted. Transmission of node IDs is needed, for example, 
to allow the root to cluster
nodes. 
To minimize communication load nodes in smaller 
clusters report violations of node constraints first, and the reporting 
order is assigned and communicated to nodes by the root that knows all 
cluster sizes. 

Since every vector $\v$ associated with a node belongs to a simplex,
it is represented by a real number not exceeding $1$. We may use
the integer part of these real numbers for transmission of integers.
There is a variety of coding and compression techniques that can be used
to transmit a set of real numbers as a single real. The discussion of 
these methods is beyond the scope of this paper. In order to be able
to compare different monitoring techniques we shall count a number of 
broadcasts, where by a broadcast we mean a single communication between
two nodes. As an illustration, below we compute the number of broadcasts
needed for one iteration of Algorithm~\ref{sec.mon.balls.al} triggered by 
violation of a node constraint. We first assume that the violator node ${\n}$
is different from the root.
\begin{enumerate}
\item
The violator node ${\n}$ notifies all other nodes (except the root) 
about the violation ($n-2$ broadcasts).
\item
Each node ${\n}$ broadcasts its vector $\v_{\n}$ to the root ($n-1$ broadcasts).
\item
The root recomputes $\delta (\r)$ and sends it to each node ($n-1$ broadcasts).
\end{enumerate}
This leads to $3(n-1)-1$ broadcasts. If the violator node ${\n}$ is the 
root itself, the number of broadcasts becomes $3(n-1)$ (at step 1 above the 
root has to make $n-1$ broadcasts).

%For this reason transmission of an integer is not counted as a message.
%The messages exchanged by nodes 
%We shell disregard broadcasting integer and consider the following scenarios:
%\underline{Each cluster is a singleton}. 
%At iteration $t$ a node $\n$ detects a violation 
%of the local constraint, i.e. $\|\v_{\n}(t)-\v_{\n}(t_j)\|\ge \delta$.

Next we turn to monitoring with clustering.
The monitoring procedure starts with each node $\n$ sending its initial vector 
$\v_{\n}(t_0)$ to the root 
$\r$ (that requires $n-1$ broadcasts). The root computes the mean 
$\d {1\over n}\sum_{\n}\v_{\n}(t_0)$ of the initial vectors, computes 
$\delta(\r)$,
and broadcasts $\delta(\r)$ to each node ($n-1$ broadcasts). 
After exchanging
\begin{equation}
\label{mc.initial}
2(n-1)
\end{equation}
broadcasts the monitoring proceeds with each node being a singleton cluster.
\begin{enumerate}
\item
As long as the inequality
$$
|\v_{\n}(t)-\v_{\n}(t_0)|<\delta(\r)\hbox{ holds true for each node }\n
%\end{equation}
$$
the nodes are silent. At the first time instance $t$ when 
the inequality is violated for at least one node $\n$, the following actions
are triggered:
\begin{enumerate}
\item
the node $\n$ (if the node itself is not the root) broadcasts its ID and 
vector 
$\v_{\n}(t)$
%$$ \Delta_{\n}=\v_{\n}(t)-\v_{\n}(t_0) $$ 
to the root ($1$ broadcast), 
\item
the root issues $n-2$ requests 
for ID and $\v_{\n}(t)$ %$\Delta$ 
to the other nodes ($n-2$ broadcasts),
\item
$n-2$ nodes report their IDs and $\v_{\n}(t)$ vectors %$\Delta$'s 
to the root 
($n-2)$ broadcasts),
\end{enumerate}
This brings the number of broadcasts to $2n-3$. If the violating node is 
the root, then this number is $2n-2$. To simplify the computations we
select the largest number $2n-2$.

At this step, and keeping in mind $(\ref{mc.initial})$,
the total number of broadcasts needed to be exchanged is 
\begin{equation}
\label{me.iterations.first}
2(n-1)+2n-2=4(n-1).
\end{equation}
\item
Next
the root recomputes $\delta (\r)$, clusters nodes, and broadcasts to each node
($n-1$ broadcasts) 
its updated local constraint $\delta (\n)$, 
the ID of its coordinator, and the reporting order. If a node is also a
coordinator, then IDs of its nodes, and coordinator reporting order are
provided to the coordinator by the root.
Keeping in mind $(\ref{me.iterations.first})$, the total number of broadcasts
right after the first root mean update and first clustering is
\begin{equation}
\label{me.cluster.first}
5(n-1).
\end{equation}
\end{enumerate}
Clusters are now formed, and we shall count the number of broadcasts 
needed to be exchanged for each of the three types of possible violations:
\begin{enumerate}
\item
\underline{A node constraint is violated in a singleton cluster}.
\begin{enumerate}
\item
the violating node $\n$ reports its ID, $\v_{\n}(t)$, %$\Delta_{\n}$, 
$W_{\n}$,
and the history vector
$\h_{\n}$ to the root 
($1$ broadcasts),
\item
the root requests all other $n-2$ nodes to provide their input 
(ID's, $\v_{\n}(t)$ vectors, %$\Delta$ vectors, 
$W_{\n}$ weights, and history vectors
$\h$, total of $n-2$ broadcasts), %{\bf ----scalars not counted--}
\item
the $n-2$ nodes report ID's, $\v_{\n}(t)$ vectors, %$\Delta$ vectors, 
$W_{\n}$ weights, and 
history vectors $\h$ to the root ($(n-2)$ broadcasts),
\item
the root recomputes the constraint $\delta(\r)$, node constraints 
$\delta(\n)$,
and reports to each node its coordinator ID,
%(if a node does not belong to a singleton cluster, total of $ns$ broadcasts, 
%where $ns$ is a number of nodes that do not belong to singletons).  
$\delta(\n)$, and the node ``reporting order." Cluster coordinators also
receive IDs of the nodes in their respective clusters ($n-1$ broadcasts).
%\item
%the root reports to each node its local constraint $\delta(\n)$ (this
%rquires additional $n-1$ broadcasts).
%\item
%the root reports $\delta(\c)$ to each coordinator $\c$ ($k$ broadcasts,
%where $k$ is the number of nonsingleton clusters).
\end{enumerate}
This leads to $3(n-1)-1$ broadcasts if the violating node is not the root,
and $3(n-1)$ broadcasts if the violation is at the root.
To compute the broadcasts we use the larger number
\begin{equation}
\label{mc.singleton}
%2(n-1)\times (d+1)-1
%2(n-1)\times d\; \hbox{\bf{--scalars not counted--}}
3(n-1).
\end{equation}
\item
\underline{A node constraint is violated in a non singleton cluster 
$\pi$ with coordinator $\c$}.
%Next consider a violations of a local constraint $\delta (\n)$ at the node 
%$\n$ that belongs to cluster $\pi$ with coordinator $\c$.
\begin{enumerate}
\item
the violator $\n$ reports its ID, $\Delta_{\n}$, and $\delta_{\n}$ 
to the coordinator $\c$ ($1$ broadcast),
\item
the coordinator $\c$ sends request for $\Delta_{\n}$ vectors 
and node constraints 
$\delta_{\n}$ for all nodes in its cluster $\pi$ other then
$\n$ and itself ($|\pi|-2$ broadcasts) %{\bf{--scalars not counted--}},
\item
the nodes broadcast their vectors $\Delta$ and constraints $\delta$ 
to the coordinator (total of $\d  \left(|\pi|-2\right)$ 
broadcasts). The total comes to $2|\pi|-3$, and this number is $2|\pi|-2$
when the violating node is the coordinator.
\end{enumerate}
The total of broadcasts needed is:
\begin{equation}
\label{mk.inclus.num.mess}
2|\pi|-2.
%(|\pi|-1)(d+1)\;\hbox{\bf{--scalars not counted--}}
\end{equation}
\item
\underline{A coordinator constraint is violated}.
First we assume the coordinator $\c$ is not the root:
\begin{enumerate}
\item
the coordinator $\c$ of cluster $\pi$ broadcasts requests to all nodes 
(except itself and the root)
to provide the root with their IDs, vectors $\v_{\n}(t)$, % vectors, $\Delta$, 
weights $W$, and
history vectors $\h$ ($n-2$ broadcasts).
%{\bf{--scalars not counted--}}
\item
$n-1$ nodes ($n-2$ nodes requested by the coordinator and the coordinator 
itself) send the requested information to the root ($n-1$ 
broadcasts).
\item
the root recomputes $\delta (\r)$, clusters nodes and provides each node 
with updated
local constraint $\delta (\n)$, the new cluster affiliation (i.e. ID of a new 
coordinator), and the node ``reporting order." Coordinators are also provided
with the IDs of their nodes 
(total of $n-1$ broadcasts).
%\item
%the root sends to each new coordinator IDs of all its nodes, and the 
%coordinator new local constraint $\delta(c)$
%(exept coordinator's own ID), 
%(total of $k$ broadcasts, where $k$ is the new number of clusters).
\end{enumerate}
This brings the number of broadcasts to
$3(n-1)-1$.
%$(n-1)\times (2d+2)-1$ %\;\hbox{\bf{--scalars not counted--}}$
If $\c$ is the root, then this number is $3(n-1)$, and this is the 
number we use to compute broadcasts
\begin{equation}
\label{mk.total.num.mess}
%(n-1)\times (2d+3).
%(n-1)\times (2d+2) %\;\hbox{\bf{--scalars not counted--}}
3(n-1).
\end{equation}
\end{enumerate}
\end{document}

