10.4.1 State of the Art

Several systems, combining principles from data management and distributed processing, have been developed over time to support different subsets of requirements that are now central to SPAs. These systems include traditional databases and data warehouses, parallel processing frameworks, active databases, continuous query systems, publish–subscribe (pub‐sub) systems, complex event processing (CEP) systems, and more recently the Map‐Reduce frameworks. A quick summary of the capabilities of these systems with respect to the streaming application characteristics defined in Section 10.3 is included in Table 10.1.

More details on the individual systems and their examples can be obtained from Ref. [9]. However, as is clear, none of these systems were truly designed to handle all requirements of SPAs. Even the recent MapReduce paradigm does not support SIMA and ATA—critical to the needs of these types of applications. As a consequence, there is an urgent need to develop more sophisticated SPSs.

10.4.2 Stream Processing Systems

While SPSs were developed by incorporating ideas from these preceding technologies, they required several advancements to the state of the art in algorithmic, analytic, and systems concepts. These advances include sophisticated and extensible programming models, allowing continuous, incremental, and adaptive algorithms and distributed, fault‐tolerant, and enterprise‐ready infrastructures or runtimes. These systems are designed to allow end‐to‐end distribution of real‐time analysis, as needed in a fog computing world. Examples of early SPSs include TelegraphCQ, STREAM, Aurora–Borealis, Gigascope, and Streambase [9]. Currently available and widely used SPSs include IBM InfoSphere Streams [3] (streams) and open‐source Storm [4] and Spark [5] platforms. These platforms are freely available for experimentation and usage in commercial, academic, and research settings. These systems have been extensively deployed in multiple domains for telecommunication call detail record analysis, patient monitoring in ICUs, social media monitoring for real‐time sentiment extraction, monitoring of large manufacturing systems (e.g., semiconductor, oil, and gas) for process control, financial services for online trading and fraud detection, environmental and natural systems monitoring, etc. Descriptions of some of these real‐world applications can be found in Ref. [9]. These systems have also been shown to scale rates of millions of data items per second, deployed across tens of thousands of processors in a distributed cluster, provide latencies of microseconds or lower, and connect with millions of distributed sensors.

While we omit a detailed description of stream processing platforms, we illustrate some of their core capabilities and constructs to support the needs of SPAs by outlining an implementation of the transportation application described in Section 10.2 in a stream programming language (SPL). In Figure 10.2, the application is shown as a flowgraph, which captures logical flow of data from one processing stage to another. Representing an application as a flowgraph allows for modular design and construction of the implementation, and as we discuss later, it allows SPSs to optimize the deployment of the application onto distributed computational infrastructures.

Each node on the processing flowgraphs in Figure 10.2 that consumes and/or produces a stream of data is labeled an operator. Individual streams carry data items or tuples that can contain structured numeric values, unstructured text, semi‐structured content such as XML, and binary blobs. In Table 10.2 we present an outline implementation in SPL [3]. Note that this flowgraph actually implements the distributed learning framework that will be discussed in more detail in Section 10.5.

composite Learner1 {
graph
stream <TPhone> PhoneStream = TCPSource()
	{param role: server; port: 12345u;}
stream <TWeather> WeatherStream = InetSource()
	{param URIList: ["http://noaa.org/xx"];}
stream <TPhone> CleanPhoneStream = Custom(PhoneStream){…}
stream <TWeather> CleanWeatherStream =
	Custom(WeatherStream){…}
stream <TFeature> FeatureStream = Join(CleanWeatherStream;
	CleanPhoneStream){…}
stream <TPrediction> P1 = MySVM(FeatureStream){…}
stream <TPrediction> P2 = MyNN(FeatureStream){…}
stream <TPrediction> Learner1Pred = MyAggregation(P1;
	P2;Feedback;Learner2Pred){…}
() as Sink2 = Export(Learner1Pred)
	{param properties: {name = "Learner1"}}
stream <TPrediction> Learner2Pred = Import()
	{param properties: {name = "Learner2"}}
stream <TFeedback> Feedback = Import()
	{param properties: {name = "Feedback"}}
}

In this code, logical composition is indicated by an operator instance stream <type> S3 = MyOp(S1;S2), where operator MyOp consumes streams S1 and S2 to produce stream S3, where type represents the type of tuples on the stream. Note that these streams may be produced and consumed on different computational resources—but that is transparent to the application developer. Systems like Streams also include multiple operators/tools that are required to build such an application. Examples include operators for:

• Data Sources and Connectors, for example, FileSource, TCPSource, ODBCSource
• Relational Processing, for example, Join, Aggregate, Functor, Sort
• Time Series Analysis, for example, Resample, Normalize, FFT, ARIMA, GMM
• Custom Extensions, for example, user‐created operators in C++/Java or wrapping for MATLAB, Python, and R code

These constructs allow for the implementation of distributed and ensemble learning techniques, such as those introduced in the learning framework, within specialized operators. We discuss this more in Section 10.5. Stream processing platforms also include special tools for geo‐spatial processing (e.g., for distance computation, map matching, speed and direction estimation, bounding box calculations) and standard mathematical processing. A more exhaustive list is available from Ref. [3].

Finally, programming constructs like the Export operator allows applications (in our case the congestion prediction application) to publish their output stream such that other applications, for example, other learners, can dynamically connect to receive these results. This construct allows for dynamic connections to be established and torn down as other learners are instantiated or choose to communicate. This is an important requirement for the learning framework in Section 10.5.

These constructs allow application developers to focus on the core algorithm design and logical flowgraph construction, while the system provides support for communication of tuples across operators, conversion of the flowgraph into a set of processes or processing elements (PEs), distribution and placement of these PEs across a distributed computation infrastructure, and finally necessary optimizations for scaling and adapting the deployed applications. We present an illustration of the process used by such systems to convert a logical flowgraph to a physical deployment in Figure 10.3.

Among the biggest strengths of using a stream processing platform is the natural scalability and efficiency provided by these systems and their support for extensibility in terms of optimization techniques for topology construction, operator fusion, operator placement, and adaptation [9]. These systems allow users to formulate and provide additional algorithms to optimize their applications based on the application, data, and resource‐specific requirements. This opens up several new research problems related to the joint optimization of algorithms and systems. For instance, consider a simple example application with two operators. Using the Streams composition language, these operators—shown as black and white boxes in Figure 10.4—can be arranged into a parallel (task parallel) or a serial (pipelined parallel) topology to perform the same task,³ as shown in Figure 10.4. This topology can also be distributed across computational resources (shown as different sized CPUs in Figure 10.4). The right choice of topology and placement depends on the resource constraints and data characteristics and needs to be dynamically adapted over time. This requires solving a joint resource optimization problem whose solution can be realized using the controls provided by SPSs. In practice, there are several such novel optimization problems that can be formulated and solved—in this joint application–system research space. For instance, Refs. [11] and [13] discuss topology construction and optimization for non trivial compositions of operators for multi‐class classification. Additionally, these systems provide support for design of novel meta‐learning and planning‐based approaches to dynamically construct, optimize, and compose topologies of operators on these systems [14].

This combination of systems and algorithms enables several other open research problems in this space of joint application–system research, especially in an online, distributed, large‐scale setting. In the next section we propose a solution to build a large‐scale online distributed ensemble learning framework that leverages the capabilities provided by SPSs (and is implemented by the code in Table 10.2) to provide solutions to the illustrative problem defined in Section 10.2.

10.5.1 State of the Art

As mentioned in Section 10.3, it is important for stream processing algorithms to be online, one pass, adaptive, and distributed to operate effectively under budget constraints and to support combinations (or ensembles) of multiple techniques. Recently, there has been research that uses the aforementioned techniques for analysis, and we include a summary of some of these approaches next. We partition our review into Ensemble methods, Diffusion adaptation, and finally frameworks for distributed learning.

10.5.2 Systematic Framework for Online Distributed Ensemble Learning

We now proceed to formalize the problem of large‐scale distributed learning from heterogeneous and dynamic data streams using the problem defined in Section 10.2. Formally, we consider a set of learners that are geographically distributed and connected via links among pairs of learners. We say that there is a link (i, j) between learners i and j if they can communicate directly with each other. In the case of our congestion application, each learner observes part of the transportation network by consuming geographically local readings from sensors and phones within a region and is linked to other learner streams via interfaces like the export–import interface described in Section 10.4.

Each learner is an ensemble of local classifiers that observes a specific set of data sources and relies on them to make local classifications, that is, partition data items into multiple classes of interest.⁴ In our application scenario, this maps to a binary classification task—predicting presence of congestion at a certain location within a certain time window. Each local classifier may be an arbitrary function (e.g., implemented using well‐known techniques such as neural networks, decision trees, etc.) that performs classification for the classes of interest. In Table 10.2 we show an implementation of this in an SPL with two local classifiers, MySVM and MyNN operators, and an aggregate MyAggregation operator. In order to simplify the discussion, we assume that each learner exploits a single local classifier, and we focus on binary classification problems, but it is possible to generalize the approach to the multi‐classifier and multi‐class cases. Each learner is also characterized by a local aggregation rule, which is adaptive.

Raw data items in our application can include sensor readings from the transportation network and user phones, as well as information about the weather. These data items are cleaned, preprocessed, and merged, and features are extracted from them (e.g., see Table 10.2), which are then sent to the geographically appropriate learner. We assume a synchronous processing model with discrete time slots. At each time slot, each learner observes a feature vector. It first exploits the local classifier to make a local prediction for that slot, and then it sends its local prediction to other learners in its neighborhood and receives local predictions from the other learners, before it finally exploits the aggregation rule to combine its local predictions and the predictions from its neighbors into a final classification.

Consider a discrete time model in which time is divided into slots, but an extension to a continuous time model is possible. At the beginning of the n‐th time slot, K multidimensional instances , , and K labels , , are drawn from an underlying and unknown joint probability distribution. Each learner i observes the instance , and its task is to predict the label (see Figure 10.6). We shall assume that a label is correlated with all the instances , . In this way, learner i’s observation can contain information about the label that learner j has to predict. We remark that this correlation is not known beforehand.

Each learner i is equipped with a local classifier that generates the local prediction based on the observed instance at time slot n. Our framework can accommodate both static pre‐trained classifiers and adaptive classifiers that learn online the parameters and configurations to adopt [39]. However, the focus of this section will not be on classifier design, for which many solutions already exist (e.g., support vector machines, decision trees, neural networks, etc.); instead, we will focus on how the learners exchange and learn how to aggregate the local predictions generated by the classifiers.

We allow the distributed learners to exchange and aggregate their local predictions through multihop communications; however, within one time slot a learner can send only a single transmission to each of its neighbors. We denote by learner j’s local prediction possessed by learner i before the aggregation at time instant n. The information is disseminated in the network as follows. First, each learner i observes and updates . Next, learner i transmits to each neighbor j the local prediction and the local predictions , for each learner such that the link (i, j) belongs to the shortest path between k and j. Hence, if transmissions are always correctly received, we have and , , where d _ij is the distance in number of hops between i and j. For instance, Figure 10.5 represents the flow of information toward learner 1 for a binary tree network assuming that transmissions are always correctly received. More generally, if transmissions can be affected by communication errors, we have , for some .

Each learner i employs a weighted majority aggregation rule to fuse the data it possesses and generates a final prediction as follows:

(10.1)

where is the sign function. In the example earlier, we use a threshold of 0 on the value of the argument to separate the two classes. In practice this threshold can be arbitrary; however this does not affect the discussion next, as the threshold corresponds to a constant shift in the argument.

In the earlier construction, learner i first aggregates all possessed predictions using the weights and then uses the sign of the fused information to output its final classification, . While weighted majority aggregation rules have been considered before in the ensemble learning literature [17– 20], there is an important distinction in Equation (10.1) that is particularly relevant to the online distributed stream mining context: since we are limiting the learners to exchange information only via links, learners receive information from other learners with delay (i.e., in general ), as a consequence different learners have different information to exploit (i.e., in general ).

Each learner i maintains a total of K weights and K local predictions, which we collect into vectors:

(10.2)

Given the weight vector , the decision rule (10.1) allows for a geometric interpretation: the homogeneous hyperplane in ℜ ^K that is orthogonal to separates the positive prediction (i.e., ) from the negative predictions (i.e., ).

We consider an online learning setting in which the true label is eventually observed by learner i. Learner i can then compare both and and use this information to update the weights it assigns to the other learners. Indeed, since we do not assume any a priori information about the statistical properties of the processes that generate the data observed by the various learners, we can only exploit the available observations and the history of past data to guide the design of the adaptation process over the network.

In summary, the sequence of events that takes place in a generic time slot n, represented in Figure 10.6, involves five phases:

1. Observation. Each learner i observes an instance at time n.
2. Information Dissemination. Learners send the local predictions they possess to their neighbors.
3. Final Prediction. Each learner i computes and outputs its final prediction .
4. Feedback. Learners can observe the true label that refers to a time slot .
5. Adaptation. If is observed, learner i updates its aggregation vector from to .

In the context of the discussed framework, it is fundamental to develop strategies for adapting the aggregation weights over time, in response to how well the learners perform. A possible approach is discussed next.

10.5.3 Online Learning of the Aggregation Weights

A possible approach to update the aggregation weights is to associate with each learner i an instantaneous loss function and minimize, with respect to the weight vector w _i , the cumulative loss given all observations up to time slot n. In the following we consider this approach, adopting an instantaneous hinge loss function [40]:

For each time instant n, we consider the one‐shot loss function

(10.3)

where the parameters and are the mis‐detection and false alarm unit costs and denotes the scalar product between w _i and . The hinge loss function is equal to 0 if the weight vector w _i allows to predict correctly the label ; otherwise the value of the loss function is proportional to the distance of from the separating hyperplane defined by w _i , multiplied by α ^MD if the prediction is but the label is 1, we refer to this type of error as a mis‐detection, or multiplied by α ^FA if the prediction is 1 but the label is , we refer to this type of error as a false alarm.

The hinge loss function gives higher importance to errors that are more difficult to correct with the current weight vector. A related albeit different approach is adopted in AdaBoost [17], in which the importance of the errors increases exponentially in the distance of the local prediction vector from the separating hyperplane. Here, however, the formulation is more general and allows for the diffusion of information across neighborhoods simultaneously, as opposed to assuming each learner has access to information from across the entire set of learners in the network.

We can then formulate a global objective for the distributed stream mining problem as that of determining the optimal weights by minimizing the cumulative loss given all observations up to time slot n:

(10.4)

where if has been observed by time instant n, otherwise .

To solve (10.4) learner i must store all previous labels and all previous local predictions of all the learners in the system, which is impractical in SPAs, where the volume of the incoming data is high and the number of learners is large. Hence, we adopt the stochastic gradient descent algorithm to incrementally approach the solution of (10.4) using only the most recently observed label. If label is observed at the end of time instant n, we obtain the following update rule for :

(10.5)

where is the prediction that learner i would have made at time instant m with the current weight vector . This construction allows a meaningful interpretation. It shows that learner i should maintain its level of confidence in its data when its decision agrees with the observed label. If disagreement occurs, then learner i needs to assess which local predictions lead to the misclassification: the weight that learner i adopts to scale the local predictions it receives from learner j is increased (by either α ^MD or α ^FA units, depending on the type of error) if the local prediction sent by j agreed with the label, otherwise is decreased.

[?] and [8] derive worst‐case upper bounds for the misclassification probability of a learner adopting the update rule 10.5. Such bounds are expressed in terms of the misclassification probabilities of two benchmarks: (i) the misclassification probability of the best local classifiers and (ii) the misclassification probability of the best linear aggregator. We remark that the best local classifiers and the best linear aggregator are not known and cannot be computed beforehand; in fact, this would require to know in advance the realization of the process that generates the instances and the labels.

The optimization problem (10.4) can also be solved within the diffusion adaptation framework, as proposed in Ref. [32]. In this framework the learners combine information with their neighbors, for example, in the combine‐then‐adapt (CTA) diffusion scheme, they first combine their weights and then adopt the stochastic gradient descent [32]. Figure 10.7 illustrates the difference between our approach and the CTA scheme.

We remark that the framework described so far requires each learner to maintain a weight (i.e., an integer value) and a local prediction (i.e., a Boolean value) for each other learner in the network. This means that the memory and computational requirements scale linearly in the number of learners K. However, notice that the aggregation rule 10.1 and the update rule 10.5 only require basic operations such as add, multiply, and compare. Moreover, if the learners have a common goals, that is, they must predict a common class label y ⁿ , it is possible to develop a scheme in which each learner keeps track only of its own local prediction and of the weight used to scale its local prediction and is responsible to update such a weight. In this scheme the learners exchange the weighted local predictions instead of the local predictions and the memory and computational requirements scale as a constant in the number of learners K. For additional details, we refer the reviewer to Ref. [8].

The framework discussed in this subsection naturally maps onto a deployment using an SPS. Each of the learners shown in Figure 10.6 maps onto the subgraph labeled learner in Figure 10.2, with the local classifiers mapping onto the shown parallel topology and the aggregation rule mapping to the fan‐in on that subgraph. As mentioned earlier, the base classifiers may be implemented using the toolkits provided by systems like Streams that include wrappers for R and MATLAB. The feedback corresponds to the delayed feedback in Figure 10.2, and the input feature vector is computed by the Pre‐proc part of the subgraph in Figure 10.2 and can include different types of spatiotemporal processing and feature extraction. Finally, the communication between the learners in Figure 10.6 is enabled by the learner information exchange connections in Figure 10.2. In summary, this online, distributed ensemble learning framework can naturally be implemented on a stream processing platform. This combination is very powerful, as it now allows the design, development, and deployment of such large‐scale complex applications much more feasible, and it also enables a range of novel signal processing, optimization, and scheduling research. We discuss some of these open problems in Section 10.6.

10.5.4 Collision Detection Application

In this subsection we apply the framework described in Sections 10.5.2 and 10.5.3 to a collision detection application in which a set of speed sensors—which are spatially distributed along a street—must detect in real‐time collisions that occur within a certain distance from them.

We consider a set of K speed sensors that are distributed along both travel directions of a street (see the left side of Figure 10.8). We focus on a generic sensor i that must detect the occurrence of collision events within z miles from its location along the corresponding travel direction, where z is a predetermined parameter. A collision event e _ℓ is characterized by an unknown starting time t _ℓ,start, when the collision occurs, and an unknown ending time t _ℓ,end, when the collision is cleared. The goal of sensor i is to detect the collision e _ℓ by the time , where T _max can be interpreted as the maximum time after the occurrence of the collision such that the information about the collision occurrence can be exploited to take better informed actions (e.g., the average time after which a collision is reported by other sources).

We divide the time into slots of length T. We write if a collision occurs at or before time instant n and is not cleared by time instant n, whereas we write to represent the absence of a collision. Figure 10.9 illustrates these notations.

At the beginning of the n‐th time slot, each speed sensor j observes a speed value , which represents the average speed value of the cars that have passed through sensor j from the beginning of the ‐th time slot until the beginning of the n‐th time slot. We consider a threshold‐based classifier:

(10.6)

where is the average speed observed by sensor j during that day of the week and time of the day, and is a threshold parameter.

If sensor j is close to sensor i, the speed value and the local prediction are correlated to the occurrence or absence of the collision events that sensor i must detect. For this reason, to detect collisions in an accurate and timely manner, the sensors must exchange their local predictions. Specifically, we denote the sensors such that sensor i precedes sensor in the direction of travel. In order to detect whether a collision has occurred within z miles from its location, sensor i requires the observations of the subsequent sensors in the direction of travel (e.g., sensors , , etc.) up to the sensor that is far from sensor i more than z miles. Hence, the information flows in the opposite direction with respect to the direction of travel. Such a scenario is represented by the right side of Figure 10.8. Notice that sensor is responsible to collect the observation from sensor and to send to sensor i both its observation and the observation of .

Figure 10.8 shows also the flow of information provided by one side of the street to the other side of the street (i.e., from sensor k to sensor i). Indeed, the fact that the observations on one side of the street are not influenced by a collision on the other side of the street can be extremely useful to assess the traffic situation and distinguish between collisions and other types of incidents. For example, the sudden decrease of the speed observed by some sensors in the considered travel direction may be a collision warning sign; however, if at the same time instants the speed observed by the sensors in the opposite travel direction decreases as well, then an incident that affect both travel directions may have occurred (e.g., it started to rain) instead of a collision.

We can formally define the flow of information represented by the right side of Figure 10.8 with a directed graph  ,⁵ where is the subset of sensors that send their local predictions to sensor i (included i itself) and is the set of links among pairs of sensors.

Now both the local classifiers and the flow of information are defined, learner i can adopt the framework described in Sections 10.5.2 and 10.5.3 to detect the occurrence of collisions within z miles from its location. Specifically, learner i maintains in memory K _i weights (K _i is the cardinality of ) that are collected in the weight vector ; it predicts adopting 10.1 and it updates the weights adopting 10.5 whenever a feedback is received. The feedback about the occurrence of the collision event e _ℓ can be provided, for example, by a driver or by a police officer, and it is in general received with delay. In Figure 10.8 such a delay is denoted by Z _ℓ .

We have evaluated the proposed framework over real‐word datasets. Specifically, we have exploited a dataset containing the speed readings of the loop sensors that are distributed along a 50 mile segment of the freeway 405 that passes through Los Angeles County and a collision dataset containing the reported collisions that occurred along the freeway 405 during the months of September, October, and November 2013. For a more detailed description of the datasets, we refer the reader to Ref. [41]. Our illustrative results show that the considered framework is able to detect more than half of the collisions occurring within a distance of 4 miles from a specific sensors while generating false alarms in the order of one false alarm every 100 predictions. The results show also that by setting the ratio α ^MD/α ^FA, mis‐detections and false alarms can be traded off.