A Distributed Online Learning Approach for Pattern Prediction over Movement Event Streams with Apache Flink Ehab Qadah Michael Mock Fraunhofer IAIS Fraunhofer IAIS Sankt Augustin, Germany Sankt Augustin, Germany Ehab.Qadah@iais.fraunhofer.de Michael.Mock@iais.fraunhofer.de Elias Alevizos Georg Fuchs NCSR "Demokritos" Fraunhofer IAIS Athens, Greece Sankt Augustin, Germany alevizos.elias@iit.demokritos.gr Georg.Fuchs@iais.fraunhofer.de ABSTRACT data with low latency is challenging, since data streams are large In this paper, we present a distributed online prediction system and distributed in nature and continuously arrive at a high rate. for user-defined patterns over multiple massive streams of move- In this paper, we present the design and implementation of an ment events, built using the general purpose stream processing online, distributed and scalable pattern prediction system over framework Apache Flink. The proposed approach is based on multiple, massive streams of events. More precisely, we consider combining probabilistic event pattern prediction models on multi- event streams related to trajectories of moving objects (i.e., ves- ple predictor nodes with a distributed online learning protocol in sels). The proposed approach is based on a novel method that order to continuously learn the parameters of a global prediction combines a distributed online prediction protocol [8, 16] with model and share them among the predictors in a communication- an event forecasting method based on Markov chains [2]. It is efficient way. Our approach enables the collaborative learning implemented on top of the Big Data framework for stream pro- between the predictors (i.e., "learn from each other"), thus the cessing Apache Flink [13]. We evaluate our proposed system over learning rate is accelerated with less data for each predictor. The real-world data streams of moving vessels, which are provided underlying model provides online predictions about when a pat- in the context of the datAcron project1 . tern (i.e., a regular expression over the event types) is expected The rest of the paper is organized as follows. We discuss the to be completed within each event stream. We describe the dis- related work and used frameworks in Section 2. In Section 3, tributed architecture of the proposed system, its implementation we describe the problem of pattern prediction, our proposed in Flink, and present experimental results over real-world event approach, and the architecture of our system. The implementa- streams related to trajectories of moving vessels. tion details on top of Flink are presented in Section 4 and the experimental results in Section 5. We conclude in Section 6. 1 INTRODUCTION 2 RELATED WORK AND BACKGROUND In recent years, technological advances have led to a growing 2.1 Related work availability of massive amounts of continuous streaming data Pattern prediction over event streams. The task of forecasting (i.e., data streams observing events) in many application domains over time-evolving streams of data can be formulated in various such as social networks [23], Internet of Things (IoT) [24], and ways and with varying assumptions. One common way to for- maritime surveillance [31]. The ability to detect and predict the malize this task is to assume that the stream is a time-series of full matches of a pattern of interest (e.g., a certain sequence of numerical values, and the goal is to forecast at each time point events), defined by a domain expert, is typically important for n the values at some future points n + 1, n + 2, etc., (or even operational decision making tasks in the respective domains. the output of some function of future values). This is the task of An event stream is an unbounded collection of time-ordered time-series forecasting [25]. Another way to formalize this task is data observations in the form of a tuple of attributes that is to view streams as sequences of events, i.e., tuples with multiple, composed of a value from finite event types along with other possibly categorical, attributes, like event type, timestamp, etc., categorical and numerical attributes. In this work, we deal with and the goal is to predict future events or patterns of events. In movement event streams. For instance, in the context of mar- this paper, we focus on this latter definition of forecasting (event itime surveillance the event stream of a moving vessel consists pattern forecasting). of spatio-temporal and kinematic information along with the A substantial body of work on event forecasting comes from vessel’s identification and its trajectory related events, based the field of temporal pattern mining where events are defined as on the automatic identification system (AIS) [28] messages that 2-tuples of the form (EventType, Timestamp). The ultimate goal are continuously sent by the vessel. Therefore, leveraging event is to extract patterns of events in the form either of association patterns prediction over real-time streams of moving vessels is rules [1] or frequent episode rules [22]. These methods have been useful to alert maritime operation managers about suspicious extended in order to be able to learn not only rules for detecting activities (e.g., fast sailing vessels near ports, or illegal fishing) event patterns but also rules for predicting events. For example, before they happen. However, processing real-time streaming in [32], a variant of association rule mining is where the goal is © 2018 Copyright held by the owner/author(s). Published in the Workshop to extract sets of event types that frequently lead to a rare, target Proceedings of the EDBT/ICDT 2018 Joint Conference (March 26, 2018, Vienna, event within a temporal window. Austria) on CEUR-WS.org (ISSN 1613-0073). Distribution of this paper is permitted under the terms of the Creative Commons license CC-by-nc-nd 4.0. 1 http://www.datacron-project.eu/ 109 In [19], a probabilistic model is presented for calculating the is treated as a special case of the streaming one (i.e., finite stream). probability of the immediately next event in the stream. This Flink’s software stack includes the DataStream and DataSet APIs is achieved by using standard frequent episode discovery algo- for processing infinite and finite data, respectively. These two rithms and combining them with Hidden Markov Models and core APIs are built on top of Flink’s core dataflow engine and mixture models. The framework of episode rules is employed provide operations on data streams or sets such as mapping, in [9] as well. The output of the proposed algorithms is a set filtering, grouping, etc. of predictive rules whose antecedent is minimal (in number of The two main data abstractions of Flink are DataStream and events) and temporally distant from the consequent. In [35] a DataSet, they represent read-only collections of data elements. set of algorithms is proposed that target batch online mining of The list of elements is bounded (i.e., finite) in DataSet, while it is sequential patterns, without maintaining exact frequency counts. unbounded (i.e., infinite) in the case of DataStream. Flink’s core As the stream is consumed, the learned patterns can be used to is a distributed streaming dataflow engine. Each Flink program test whether a prefix matches the last events seen in the stream, is represented by a data-flow graph (i.e., directed acyclic graph - indicating a possibility of occurrence for events that belong to DAG) that gets executed by Flink’s dataflow engine [6]. The data the suffix of the rule. flow graphs are composed of stateful operators and intermediate Event forecasting has also attracted some attention from the data stream partitions. The execution of each operator is handled filed of Complex Event Processing (see [7] for a review of Com- by multiple parallel instances whose number is determined by plex Event Processing). One such early approach is presented the parallelism level. Each parallel operator instance is executed in [26]. Complex event patterns are converted to automata, and in an independent task slot on a machine within a cluster of subsequently, Markov chains are used in order to estimate when computers [13]. a pattern is expected to be fully matched. A similar approach Apache Kafka. Apache Kafka is a scalable, fault-tolerant, and is presented in [2], where again automata and Markov chains distributed streaming framework/messaging system [11]. It al- are employed in order to provide (future) time intervals during lows to publish and subscribe to arbitrary data streams, which which a match is expected with a probability above a confidence are managed in different categories (i.e., topics) and partitioned threshold. in the Kafka cluster. The Kafka Producer API provides the abil- Distributed Online Learning. In recent years, the problem of ity to publish a stream of messages to a topic. These messages distributed online learning has received increased attention and can then be consumed by applications, using the Consumer API has been studied in [8, 16, 18, 33, 34]. A distributed online mini- that allows them to read the published data stream in the Kafka batch prediction approach over multiple data streams has been cluster. In addition, the streams of messages are distributed and proposed in [8]. This approach is based on a static synchroniza- load balanced between the multiple receivers within the same tion method. The learners periodically communicate their local consumer group for the sake of scalability. models with a central coordinator unit after consuming a fixed number of input samples/events (i.e., batch size b), in order to 3 SYSTEM OVERVIEW create a global model and share it between all learners. This work 3.1 Pattern prediction on a single stream has been extended in [16] by introducing a dynamic synchroniza- For our work presented in this paper, we use the approach pre- tion scheme that reduces the required communication overhead. sented in [2]. For the sake of self-containment, we briefly describe It can do so by making the local learners communicate their mod- this approach in the following, first assuming that only a single els only if they diverge from a reference model. In this work, we stream is consumed and then adjusting for the case of multiple employ this protocol with event patterns prediction models over streams. We follow the terminology of [3, 21, 35] to formalize multiple event streams. the problem we tackle. 2.2 Technological Background 3.1.1 Problem formulation. We define an input event and a In the last years, many systems for large-scale and distributed stream of input events as follows: stream processing have been proposed, including Spark Stream- Definition 3.1. Each event is defined as a tuple of attributes ing [12], Apache Storm [14] and Apache Flink [13]. These frame- ei = (id, type, τ , a 1 , a 2 ....., an ), where type is the event type at- works can ingest and process real-time data streams, published tribute that takes a value from a set of finite event types/symbols from different distributed message queuing platforms, such as Σ, τ represents the time when the event tuple was created, the Apache Kafka [11] or Amazon Kinesis [5]. In this work, we im- a 1 , a 2 , ..., an are spatial or other contextual features (e.g., speed); plemented our system in Flink and Kafka (see Section 4). these features are varying from one application domain to an- In the datAcron project, the Flink streaming processing engine other. The attribute id is a unique identifier that connects the has been chosen as a primary platform for supporting the stream- event tuple to an associated domain object. ing operations, based on an internal comparative evaluation of several streaming platforms. Hence, we used it to implement our Definition 3.2. A stream s = ⟨e 1 , e 3 , ..., et , ...⟩ is a time-ordered system. A predecessor distributed online learning framework has sequence of events. already been implemented in the FERARI project [10] based on A user-defined pattern P is given in the form of a regular Apache Storm. expression (i.e., using operators for sequence, disjunction, and iteration) over Σ (i.e., event types) [2]. More formally, a pattern Apache Flink. Apache Flink is an open source project that is given through the following grammar: provides a large-scale, distributed, and stateful stream processing platform [6]. Flink is one of the most recent and pioneering Big Definition 3.3. P := E | P1 ; P2 |P1 ∨ P2 | P1∗ , where E ∈ Σ is Data processing frameworks. It provides processing models for a constant event type. ; stands for sequence, ∨ for disjunction both streaming and batch data, where the batch processing model and ∗ for Kleene − ∗. The pattern P := E is matched by reading 110 an event ei iff ei .type = E. The other cases are matched as in standard automata theory. a d a The problem at hand may then be stated as follows: given a 1 a a a stream s of low-level events and a pattern P, the goal is to a b b 5 b estimate at each new event arrival the number of future events that we will need to wait for until the pattern is satisfied (and 2 b c c a therefore a full match is detected). start 0 c b b 6 bd 3.1.2 Proposed approach. As a first step, event patterns are c d d converted to deterministic finite automata (DFA) through stan- 3 c d 4 dard conversion algorithms [15]. As an example, see Figure 1a for the DFA of the simple sequential pattern P = a; d; c and an alpha- d bet Σ = {a, b, c, d} (note that the DFA has no dead states since d c we need to handle streams and not strings). The next step is to c derive a Markov chain that will be able to provide a probabilistic (a) DFAΣ∗ ;P description of the DFA’s run-time behavior. Towards this goal, we use Pattern Markov Chains, as was P(d |a) P(a|a) proposed in [27]. Under the assumption that the input stream is generated by an m-order Markov source, and after performing a P(a|d) transformation step on the initial DFA to handle the mt h order 1 P(a|b) 5 P(b |b) (see [27] for more details). It can be shown that there is a direct P(b|d) mapping of the states of the DFA to states of a Markov chain and P(b |a) 2 P(c |d) the transitions of the DFA to transitions of the Markov chain. The transition probabilities are then conditional probabilities on P(a|c) P(a|d) P(d |d) the event types. 6 P(c |a) P(b |d) We call such a derived Markov chain a Pattern Markov Chain P(c |b) P(d |b) (PMC) of order m and denote by PMC m , where P is the initial 1 P P(b |c) pattern and m the assumed order. As an example, see Figure 1b, 3 4 which depicts the PMC of order 1 for the generated DFA of P(d |c) Figure 1a. P(c |c) P(d |d) P(c |d) 1 (b) PMCP After constructing a PMC, we can use it to calculate the so- called waiting-time distributions. Given a specific state of the Figure 1: DFA and PMC for P = a; d; c with Σ = {a, b, c, d}, PMC, a waiting-time distribution gives us the probability of reach- and order m = 1. ing a set of absorbing states in n transition from now (absorbing states are states with self-loops and probability equal to 1.0). By mapping the final states of the initial DFA to absorbing states of the PMC (see again Figure 1). Therefore, we can calculate the probability of reaching a final state, or, in other words, of detect- ing a full match of the original regular expression in n events from now. In order to estimate the final forecasts, another step is required, since our aim is not to provide a single future point with the high- est probability but an interval. Predictions are given in the form of intervals, as I = (start, end). The meaning of such an interval is that the DFA is expected to reach a final state sometime in the future between the start and end with probability at least some constant threshold θ f c (provided by the user). These intervals are estimated by a single-pass algorithm that scans a waiting-time distribution and finds the smallest (in terms of length) interval (a) Waiting-time distribution. (b) Prediction intervals. that has probability exceeds this threshold (θ f c ). For example, Figure 2a shows the waiting-time distributions for the non-final Figure 2: Example of how prediction intervals are pro- states of the DFA in Figure 1, and the computed prediction inter- duced. P = a; d; c, Σ = {a, b, c, d}, m = 1, θ fc = 0.5. vals are depicted in Figure 2b. The method described above assumes that we know the (pos- sibly conditional) occurrence probabilities of the various event types appearing in a stream (as would be the case with syntheti- part of the stream to obtain the maximum-likelihood estimators cally generated streams). However, this is not always the case in for the transition probabilities [4]. If Π is the transition matrix of real-world situations. Therefore, it is crucial for a system imple- a Markov chain with a set of states Q, πi, j the transition probabil- menting this method to have the capability to learn the values of ity from state i to state j, ni, j the number of observed transitions the PMC’s transition matrix. One way to do this is to use some from state i to state j, then the maximum likelihood estimator 111 for πi, j is given by: computed from the already received models less or equal than ni, j ni, j the divergence threshold ∆. Finally, the aggregated model fˆ is π̂i, j = Í = sent back to the predictor nodes that sent their models after the n k ∈Q i,k ni violation or have been queried by the coordinator. Executing this learning step on a single node might require a vast amount of time until we arrive at a sufficiently good model. Algorithm 1: Communication-efficient Distributed Online In this paper, we present a distributed method for learning the Learning Protocol transition probability matrix. Predictor node ni : at observing event e j 3.2 Pattern prediction on multiple streams update the prediction model parameters fi and provide a 3.2.1 Problem formulation. Let O = {o 1 , ..., ok } be a set of K prediction service ; objects (i.e., moving objects) and S = {s 1 , ..., sk } a set of real-time if j mod b = 0 and ∥ fi − fr ∥ 2 > ∆ then streams of events, where si is generated by the object oi . Let P be send fi to the Coordinator (violation) ; a user-defined pattern which we want to apply to every stream si , i.e., each object will have its own DFA. Coordinator: The setting that is considered in this work is then described receive local models with violation B = { fi }i=1 m ; in the following: we have K input event streams S and a system 1 Í ˆ 2 while |B| , k and |B | fi ∈Π ∥ fi − f ∥ > ∆ do consisting of K distributed predictor nodes n 1 , n 2 ..., nk , each of add other nodes that have not reported violation for which consumes an input event stream si ∈ S. The goal is to their models B ← { fl : fl < B and l ∈ [k]} ; provide timely predictions and be able to do this at large-scale. receive models from nodes in B; Each node ni handles a single event stream si associated with a moving object oi ∈ O. In addition, it maintains a local prediction compute a new global model fˆ ; model fi for the user-defined pattern P. The fi model provides send fˆ to all the predictors in B and set f 1 . . . fm = fˆ; the online prediction about the future full match of the pattern if |B| = k then P in si for each new arriving event tuple. set a new reference model fr ← fˆ ; In short, we have multiple running instances of an online prediction algorithm on distributed nodes for multiple input event streams. More specifically, the input to our system consists This protocol was introduced for linear models, and has been of massive streams of events that describe trajectories of moving extended to handle kernelized online learning models [17]. We vessels in the context of maritime surveillance, where there is also employ this protocol for the pattern prediction model, which one predictor node for each vessel’s event stream. is internally based on the PMC PMC m P . This allows the distributed PMC m predictors for multiple event streams to synchronize their 3.2.2 The proposed approach. We designed and developed a P models (i.e., the transition probability matrix of each predictor) scalable and distributed pattern prediction system over a massive within the system in a communication-efficient manner. input event streams of moving objects. As the base prediction We propose a synchronization operation for the parameters of model, we use the PMC forecasting method [2]. Moreover, we pro- the models (fi = Πi : i ∈ [k]) of the k distributed PMC predictors. pose to enable the information exchange between the distributed The operation is based on distributing the maximum-likelihood predictors/learners of the input event streams, by adapting the estimation [4] for the transition probabilities of the underlying distributed online prediction protocol of [16] to synchronize the PMC m models described by: prediction models, i.e., the transitions probabilities matrix of the P Í PMC predictors. k ∈K nk,i, j Algorithm 1 presents the distributed online prediction protocol π̂i, j = Í Í k ∈K l ∈L nk,i,l by dynamic model synchronization on both the predictor nodes Moreover, we measure the divergence of local models from and the coordinator. We refer to the PMC’s transition matrix Πi the reference model ∥ fk − fr ∥ 2 by calculating the sum of square on predictor node ni by fi . That is, when a predictor ni : i ∈ [k] difference between the transition probabilities Πi and Πr : observes an event e j it revises its internal model state (i.e., fi ) and Õ provides a prediction report. Then it checks the local conditions ∥ fk − fr ∥ 2 = (π̂k i, j − π̂r i, j)2 (batch size b and local model divergence from a reference model i, j fr ) to decide whether there is a need to synchronize its local In general, our approach relies on enabling the collaborative model with the coordinator [or not]. fr is maintained in the learning among the distributed predictors. Each predictor node predictor node as a copy of the last computed aggregated model receives a stream of events related to a distinct moving object, fˆ from the previous full synchronization step, which is shared and the central coordinator is responsible for synchronizing their between all local predictors/learners. By monitoring the local prediction models using the synchronization operation. Moreover, condition ∥ fi − fr ∥ 2 > ∆ on all local predictors, we have a the predictors they only need to share the parameters of their guarantee that if none of the local conditions is violated, the models, not the consumed event streams. Í divergence (i.e., variance of local models δ (f ) = k1 kj=1 ∥ fi − fˆ ∥ 2 ) We assume that the underlying event streams belong to the does not exceed the threshold ∆ [16]. same distribution and share the same behavior (e.g., mobility On the other hand, the coordinator receives the prediction patterns). We claim this assumption is reasonable in many appli- models from the predictor nodes that requested for model syn- cation domains: for instance, in the context of maritime surveil- chronization (violation). Then it tries to keep incrementally query- lance, vessels travel through standard routes, defined by the ing other nodes for their local prediction models until reaching International Maritime Organization (IMO). Additionally, vessels out all nodes, or the variance of the aggregated model fˆ that is have similar mobility patterns in specific areas such as moving 112 with low speed and multiple turns near the ports [20, 29]. That al- 4 IMPLEMENTATION DETAILS lows our system to construct a coherent global prediction model In this section, we briefly describe in detail the implementation of dynamically for all input event streams based on merging their our system on top of Apache Flink and Apache Kafka frameworks. local prediction models. Each of the three sub-modules, described in Section 3.3, have been implemented as Flink operations over the Kafka events 3.3 Distributed architecture stream. Our system consumes as an input2 an aggregated stream of events Pre-processing and Prediction Operators. Listing 1 shows coming from a large number of moving objects, which is con- how the main workflow of the system is implemented as Flink tinuously collected and fed into the system. It allows users to data flow program. register a pattern P to be monitored over each event stream of The system ingests the input events stream from a Kafka a moving object. The output stream consists of original input cluster that is mapped to a DataStream of events, which is then events and predictions of full matches of P, displayed to the end processed by an EventTuplesMapper to create tuples of (id, event), users. Figure 3 presents the overview of our system architecture where the id is associated to the identifier of the moving object. and its main components. To handle events coming in out of order in a certain margin, the stream of event tuples is processed by a TimestampAssigner, it assigns the timestamps for the input events based on the extracted creation time. Afterwards, an ordered stream of event tuples is generated using a process function EventSorter. DataStream eventsStream = env.addSource(kafkaConsumer); // Create tuples (id,event) and assign time stamps DataStream> eventTuplesStream = inputEventsStream.map(new EventTuplesMapper()) .assignTimestampsAndWatermarks(new EventTimeAssigner()); // Create the ordered keyed stream orderedEventsStream = eventsStream.keyBy(0).process(new EventSorter()).keyBy(0); Figure 3: System Architecture. // Consume the events by the predictors LocalPredictorNode predictorNode =new LocalPredictorNode(P); DataStream processedEventsStream = The system is composed of three processing units: (i) pre-pro- orderedEventsStream.map(predictorNode); cessing operators that receive the input event stream and per- form filtering and ordering operations, before partitioning the Listing 1: Flink pipeline for local predictors workflow input event stream to multiple event streams based on the asso- ciated moving object (ii) predictor nodes (learners), which are The ordered stream is then transformed to a keyedEventsStream responsible for maintaining a prediction model for the input by partitioning it, based on the ids values, using a keyBy oper- event streams. Each prediction node is configured to handle an ation. A local predictor node in a distributed environment is event stream from the same moving object, in order to provide represented by a map function over the keyedEventsStream. Each online predictions for a predefined pattern P (iii) a coordinator parallel instance of the map operator (predictor) always pro- node that communicates through Kafka stream channels with cesses all events of the same moving object (i.e., equivalent id), the predictors to realize the distributed online learning protocol. and maintains a bounded prediction model (i.e., PMC m predictor) P It builds a global prediction model, based on the received local using the Flink’s Keyed State 3 . The output streams of the moving models, and then shares it among the predictors. objects from the parallel instances of the predictor map functions Our distributed system consists of multiple pre-processing op- are sent to a new Kafka stream (i.e., same topic name). They then erators, prediction nodes, and a central coordinator node. These can be processed by other components like visualization or users units run concurrently and are arranged as a data processing notifier. pipeline, depicted in Figure 3. We leverage Apache Kafka as a Moreover, the implementation of the predictor map function in- messaging platform to ingest the input event streams and to cludes the communication with coordinator using Kafka streams. publish the resulting streams. Also, it is used as the communica- At the beginning of the execution, it sends a registration request tion channel between the predictor nodes and the coordinator. to the coordinator. Also at the run-time, it sends its local pre- Apache Flink is employed to execute the system’s distributed diction model as synchronization request, or as a response for a processing units over the input event streams: the pre-processing resolution request from the coordinator. These communication operators, the prediction units, and the coordinator node. Our messages are published into different Kafka topics as depicted in system architecture can be modeled as a logical network of pro- Table 1. cessing nodes, organized in the form of a DAG, inspired by the Coordinator. It manages the distributed online learning pro- Flink runtime dataflow programs [6]. tocol operations, which is also implemented as Flink program. The coordinator receives messages from the local predictors 2 In practice, the aggregated input events stream is composed of multiple event 3 Keyed State in Flink: https://ci.apache.org/projects/flink/flink-docs-release- streams (partitions) from a set of moving objects. 1.3/dev/stream/state.html#kayed-state 113 Table 1: Messages to Kafka topics mapping. and 32GB RAM. We used Apache Flink v1.3.2 and Apache Kafka v0.10.2.1 for our tests. Message Kafka Topic Evaluation criteria. Our goal is to evaluate our distributed pat- RegisterNode, LocalToCoordinatorTopicId tern prediction system, which enables the synchronization of RequestSync, and prediction models (i.e., PMC models) on the distributed predic- ResolutionAnswer tor nodes. Our proposed system can operate in three different modes of protocols/schemes of models synchronization: (i) static CoordinatorSync and CoordinatorToLocalTopicId scheme based on synchronizing the prediction models periodi- RequestResolution cally every b of input events in each stream, (ii) continuous, full synchronization for each incoming event (hypothetical), (iii) dy- namic synchronization protocol based on making the predictors through a Kafka Stream of a topic named "LocalToCoordinator- communicate their local prediction models periodically but only TopicId". It is implemented as a single map function over the under condition that the divergence of the local models from a messages stream, by setting the parallelism level of the Flink pro- reference model exceeds a variance threshold ∆ (recommended). gram to "1". Increasing the parallelism will scale up the number We compare our proposed system against the isolated predic- of parallel coordinator instances, for example, in order to handle tion mode, in which models are computed on single streams only, different groupings of the input event streams. The map oper- and compare the predictive performance in terms of : ator of the coordinator handles three message types from the # of correct predictions predictors: (i) RegisterNode that contains a registration request (i) Precision = # of total predictions is the fraction of the pro- for a new predictor node, (ii) RequestSync to receive a local duced predictions that are correct. For each new event in model after violation, (iii) ResolutionAnswer to receive a reso- the stream, the predictor provides a prediction interval lution response from a local predictor node. In addition, it sends where the full match of the pattern might occur. Thus, the CoordinatorSync messages for all predictors after creating a predictions are temporarily stored until a full match is new global prediction model, or RequestResolution to a ask detected. At that point, all stored prediction intervals are the local predictors for their prediction models. evaluated by considering those intervals where the full match occurred within as correct. 5 EMPIRICAL EVALUATION (ii) Spread= end(I ) − start(I ) is the width of the prediction In this section, we evaluate our proposed system by analyzing interval I , which represents the number of events between the predictive performance and communication complexity us- the start and the end of I . ing real-world event streams provided by the datAcron project Moreover, we study the communication cost by measuring the in the context of maritime monitoring. The used event streams cumulative communication that captures the number of messages, describe critical points (i.e., synopses) of moving vessels trajecto- which are required to perform the distributed online learning ries, which are derived from raw AIS messages as described in modes to synchronize the prediction models. Next, we present [30]. In particular, for our evaluation experiments we used a data the experimental results for the patterns P1 = Sailinд with an set of synopses that contains 4, 684, 444 critical points of 5055 order of m = 2, and P2 =changeInHeading; gapStart; gapEnd; vessels sailing in the Atlantic Ocean during the period from 1 changeInHeading with first order m = 1. All experiments are October 2015 to 31 March 2016. performed with setting the batch size to 100 (b = 100), the vari- We used the synopses data set to generate a simulated stream ance threshold of 2 (∆ = 2), 80% as PMC prediction threshold of event tuples i.e., (id, timestamp, longitude, latitude, annotation, (θ f c = 80%), and 200 for the maximum spread. speed, heading), which are processed by the system to attach an ex- Experimental results. Figure 4 depicts the average precision tra attribute type that represents the event value, where type ∈ Σ, scores of predictions models (one prediction model per vessel) and Σ = Σ1 ={VerySlow, Slow, Moving, Sailing, Stopping}, which is of all synchronization modes for the first pattern P1 = Sailinд, based on a discretization of the speed values. That is, Σ1 includes namely, isolated without synchronization, continuous (full-sync), a simple derived event types based on the speed value that can static, and our recommended approach based on the dynamic be used over streams of raw AIS or critical points. Or Σ = Σ2 = synchronization scheme. It can be clearly seen that all methods {stopStart, stopEnd, changeInSpeedStart, changeInSpeedEnd, slow- of distributed learning outperform the isolated prediction models. MotionStart, slowMotionEnd, gapStart, gapEnd, changeInHeading}, The hypothetical method of full continuous synchronization has which is derived based on the values of the annotation attribute the highest precision rates, while the static and dynamic syn- that encodes the extracted trajectory movement events [30]. Σ2 chronization schemes have close precision scores. Consequently, represents the set of possible mobility changes in the vessel’s dynamic synchronization is not much weaker than the static trajectory [30], each critical point has at least one event. Where synchronization, but requires much less communication, as ex- in the case of multiple values, we generate duplicate points each plained below. of which corresponding to one event in the same order of Σ2 . Figure 5 provides the amount of the accumulated communica- In our experiments, we monitor a pattern P1 = Sailinд with tion that is required by the three modes of the distributed online Σ1 that detects when the vessel is underway (sailing). Likewise, learning, while the isolated approach does not require any com- we test a second pattern P2 =changeInHeading; gapStart; gapEnd; munication between the predictors. These results are shown for changeInHeading with Σ2 that describes a potential illegal fishing P1 . As expected, a larger amount of communication is required activity [2]. for the continuous synchronization comparing to the static and Experimental setup. We ran our experiments on single-node dynamic approaches. Also, it can be seen that we can reduce the standalone Flink cluster deployed on an Ubuntu Server 17.04 communication overhead by applying the dynamic synchroniza- with Intel(R) Core(TM) i7-7700 CPU @ 3.60GHz X 8 processors tion protocol (a reduction by a factor of 100) comparing to the 114 Figure 4: Precision scores with respect to the number of Figure 6: Average spread value for P1 . input events over time for P1 . static synchronization scheme, even with a small variance thresh- old ∆ = 2. Furthermore, the dynamic protocol is still preserving a close predictive performance to the static one (see Figure 4). Therefore, we will only consider the dynamic synchronization and the isolated approach in the evaluation of the second pattern. Figure 7: Precision scores of P2 for PLEASURE CRAFT ves- sels. In Figure 4, we also noted that the precision is going down in a first phase and stabilizes then. This seems to be counter-intuitive, as the models should improve when getting more data up to a certain point. For explanation, we have investigated the effect of the distributed synchronization of the prediction models on the average spread value, Figure 6 shows the spread results for Figure 5: Cumulative communication with respect to the all approaches. It can be seen that the spread is higher for the number of input events over time for P1 . distributed learning based methods comparing to the isolated approach. Furthermore, the average spread is decreasing over time until convergence, as result of confidence increase in the models. This may explain the drop in the precision scores from 115 the beginning until reaching the convergence. We will investigate [7] Gianpaolo Cugola and Alessandro Margara. 2012. Processing Flows of Infor- further in the interrelation between precision and spread in future mation: From Data Stream to Complex Event Processing. ACM Comput. Surv. 44, 3, Article 15 (June 2012), 62 pages. https://doi.org/10.1145/2187671.2187677 work. [8] Ofer Dekel, Ran Gilad-Bachrach, Ohad Shamir, and Lin Xiao. 2012. Optimal For the second, more complex pattern (P2 ), we have found that distributed online prediction using mini-batches. Journal of Machine Learning Research 13, Jan (2012), 165–202. the precision was worse for a distributed model generated over all [9] Lina Fahed, Armelle Brun, and Anne Boyer. 2014. Efficient Discovery of vessels than in the model created for each vessel in isolation. This Episode Rules with a Minimal Antecedent and a Distant Consequent. In indicates that there is no global model describing the behavior of Knowledge Discovery, Knowledge Engineering and Knowledge Management. Springer. all models consistently. However, when looking at specific groups [10] Ioannis Flouris, Vasiliki Manikaki, Nikos Giatrakos, Antonios Deligiannakis, of vessels, we achieved an improvement in terms of precision. Minos Garofalakis, Michael Mock, Sebastian Bothe, Inna Skarbovsky, Fabiana As initial experiment, we only enable the synchronization of the Fournier, Marko Stajcer, et al. 2016. FERARI: A Prototype for Complex Event Processing over Streaming Multi-cloud Platforms. In Proceedings of the 2016 prediction models associated with vessels that belong to the same International Conference on Management of Data. ACM, 2093–2096. vessel class. Currently, this change is technically performed by [11] The Apache Software Foundation. 2012. Apache Kafka. https://kafka.apache. org/. (2012). an extra filter step that passes only one type of vessels, while [12] The Apache Software Foundation. 2013. Apache Spark Streaming. http: multiple runs of the system are required for all vessel types. For //spark.apache.org/streaming/. (2013). example, Figure 7 shows the precision scores for vessels of class [13] The Apache Software Foundation. 2014. Apache Flink. https://flink.apache. org/. (2014). PLEASURE CRAFT. An interesting observation is that the dynamic [14] The Apache Software Foundation. 2014. Apache Storm. http://storm.apache. synchronization approach still has higher precision scores than org/. (2014). the isolated approach. This case might seem to contradict of [15] John E Hopcroft, Rajeev Motwani, and Jeffrey D Ullman. 2006. Automata theory, languages, and computation. International Edition 24 (2006). our assumption that the input event streams belong to the same [16] Michael Kamp, Mario Boley, Daniel Keren, Assaf Schuster, and Izchak Sharf- distribution and share the same behavior, but it actually follows man. 2014. Communication-efficient distributed online prediction by dynamic model synchronization. In Joint European Conference on Machine Learning the same assumption but between the predictors of vessels within and Knowledge Discovery in Databases. Springer, 623–639. the same type group. We will further investigate the effect of [17] Michael Kamp, Sebastian Bothe, Mario Boley, and Michael Mock. 2016. groupings and more patterns in future work. Communication-Efficient Distributed Online Learning with Kernels. In Joint European Conference on Machine Learning and Knowledge Discovery in Databases. Springer, 805–819. 6 CONCLUSION [18] John Langford, Alex J Smola, and Martin Zinkevich. 2009. Slow learners are fast. Advances in Neural Information Processing Systems 22 (2009), 2331–2339. In this paper, we have presented a system that provides a dis- [19] Srivatsan Laxman, Vikram Tankasali, and Ryen W. White. 2008. Stream tributed pattern prediction over multiple large-scale event streams Prediction Using a Generative Model Based on Frequent Episodes in Event Sequences. In ACM SIGKDD. of moving objects (vessels). The system uses the event forecast- [20] Bo Liu, Erico N de Souza, Stan Matwin, and Marcin Sydow. 2014. Knowledge- ing with Pattern Markov Chain (PMC) [2] as the base prediction based clustering of ship trajectories using density-based approach. In Big Data (Big Data), 2014 IEEE International Conference on. IEEE, 603–608. model on each event stream, and it applies the protocol for dis- [21] David Luckham. 2008. The power of events: An introduction to complex event tributed online prediction [16] to exchange information between processing in distributed enterprise systems. In International Workshop on the prediction models over multiple input event streams. Our Rules and Rule Markup Languages for the Semantic Web. Springer, 3–3. [22] Heikki Mannila, Hannu Toivonen, and A. Inkeri Verkamo. 1997. Discovery of proposed system has been implemented using Apache Flink and Frequent Episodes in Event Sequences. Data Mining and Knowledge Discovery Apache Kafka. In order to show the usefulness and effectiveness (1997). of our approach, we empirically tested it against large real-world [23] Michael Mathioudakis and Nick Koudas. 2010. Twittermonitor: trend detection over the twitter stream. In Proceedings of the 2010 ACM SIGMOD International event streams related to trajectories of moving vessels. Conference on Management of data. ACM, 1155–1158. As future work, we will address the open issues emerging [24] Daniele Miorandi, Sabrina Sicari, Francesco De Pellegrini, and Imrich Chlam- tac. 2012. Internet of things: Vision, applications and research challenges. Ad from the current findings. Firstly, we will study the interrelation Hoc Networks 10, 7 (2012), 1497–1516. between precision and spread scores by validating the approach [25] Douglas C Montgomery, Cheryl L Jennings, and Murat Kulahci. 2015. Intro- over synthetic event streams. Secondly, we will investigate the duction to time series analysis and forecasting. Wiley. [26] Vinod Muthusamy, Haifeng Liu, and Hans-Arno Jacobsen. 2010. Predictive effect of grouping the input event streams on the predictive Publish/Subscribe Matching. In DEBS. ACM. performance of our method. [27] Grégory Nuel. 2008. Pattern Markov Chains: Optimal Markov Chain Embed- ding through Deterministic Finite Automata. Journal of Applied Probability (2008). ACKNOWLEDGMENTS [28] International Maritime Organization. 2001. Automatic identification systems. http://www.imo.org/OurWork/Safety/Navigation/Pages/AIS.aspx. (2001). This work was supported by EU Horizon 2020 datAcron project [29] Giuliana Pallotta, Michele Vespe, and Karna Bryan. 2013. Vessel pattern (grant agreement No 687591). knowledge discovery from AIS data: A framework for anomaly detection and route prediction. Entropy 15, 6 (2013), 2218–2245. [30] Kostas Patroumpas, Elias Alevizos, Alexander Artikis, Marios Vodas, Nikos REFERENCES Pelekis, and Yannis Theodoridis. 2017. Online event recognition from moving vessel trajectories. GeoInformatica 21, 2 (2017), 389–427. [1] Rakesh Agrawal, Tomasz Imieliński, and Arun Swami. 1993. Mining Associa- [31] Kostas Patroumpas, Alexander Artikis, Nikos Katzouris, Marios Vodas, Yan- tion Rules Between Sets of Items in Large Databases. In ACM SIGMOD. nis Theodoridis, and Nikos Pelekis. 2015. Event Recognition for Maritime [2] Elias Alevizos, Alexander Artikis, and George Paliouras. 2017. Event Forecast- Surveillance.. In EDBT. 629–640. ing with Pattern Markov Chains. In Proceedings of the 11th ACM International [32] R. Vilalta and Sheng Ma. 2002. Predicting rare events in temporal domains. In Conference on Distributed and Event-based Systems. ACM, 146–157. ICDM. [3] Elias Alevizos, Anastasios Skarlatidis, Alexander Artikis, and Georgios [33] Lin Xiao. 2010. Dual averaging methods for regularized stochastic learning Paliouras. 2015. Complex event recognition under uncertainty: A short sur- and online optimization. Journal of Machine Learning Research 11, Oct (2010), vey. Event Processing, Forecasting and Decision-Making in the Big Data Era 2543–2596. (EPForDM) (2015), 97–103. [34] Feng Yan, Shreyas Sundaram, SVN Vishwanathan, and Yuan Qi. 2013. Dis- [4] Theodore W Anderson and Leo A Goodman. 1957. Statistical inference about tributed autonomous online learning: Regrets and intrinsic privacy-preserving Markov chains. The Annals of Mathematical Statistics (1957), 89–110. properties. IEEE Transactions on Knowledge and Data Engineering 25, 11 (2013), [5] Amazon Web Services (AWS). 2013. Amazon Kinesis. https://aws.amazon. 2483–2493. com/de/kinesis/. (2013). [35] Cheng Zhou, Boris Cule, and Bart Goethals. 2015. A pattern based predictor [6] Paris Carbone, Asterios Katsifodimos, Stephan Ewen, Volker Markl, Seif Haridi, for event streams. Expert Systems with Applications (2015). and Kostas Tzoumas. 2015. Apache flink: Stream and batch processing in a single engine. Bulletin of the IEEE Computer Society Technical Committee on Data Engineering 36, 4 (2015). 116