Incremental Query Processing on Big Data Streams

This paper addresses online query processing for large-scale, incremental data analysis on a distributed stream processing engine (DSPE). Our goal is to convert any SQL-like query to an incremental DSPE program automatically. In contrast to other approaches, we derive incremental programs that return accurate results, not approximate answers. This is accomplished by retaining a minimal state during the query evaluation lifetime and by using incremental evaluation techniques to return an accurate snapshot answer at each time interval that depends on the current state and the latest batches of data. Our methods can handle many forms of queries on nested data collections, including iterative and nested queries, group-by with aggregation, and equi-joins. Finally, we report on a prototype implementation of our framework, called MRQL Streaming, running on top of Spark and we experimentally validate the effectiveness of our methods.

INTRODUCTION
In recent years, large volumes of data are being generated, analyzed, and used at an unprecedented scale and rate. Data analysis tools that process these data are typically batch programs that need to work on the complete datasets, thus repeating the computation on existing data when new data are added to the datasets. Consequently, batch processing may be prohibitively expensive for Big Data that change frequently. For example, the Web is evolving at an enormous rate with new Web pages, content, and links added daily. Web graph analysis tools, such as PageRank, which are used extensively by search engines, need to recompute their Web graph measures very frequently since they become outdated very fast. There is a recent interest in incremental Big Data analysis, where data are analyzed in incremental fashion, so that existing results on current data are reused and merged with the results of processing the new data. Incremental data processing can generally achieve better performance and may require less memory than batch processing for many data analysis tasks. It can also be used for analyzing Big Data incrementally, in batches that can fit in memory. Consequently, incremental data processing can also be useful to stream-based applications that need to process continuous streams of data in real-time with low latency, which is not feasible with existing batch analysis tools. For example, the Map-Reduce framework [14], which was designed for batch processing, is ill-suited for certain Big Data workloads, such as real-time analytics, continuous queries, and iterative algorithms. New alternative frameworks have emerged that address the inherent limitations of the Map-Reduce model and perform better for a wider spectrum of workloads. Currently, among them, the most promising frameworks that seem to be good alternatives to Map-Reduce while addressing its drawbacks are Google's Pregel [29], Apache Spark [40], and Apache Flink [20], which are in-memory distributed computing systems. There are also quite a few emerging distributed stream processing engines (DSPEs) that realize online, low-latency data processing with a series of batch computations at small time intervals, using a continuous streaming system that processes data as they arrive and emits continuous results. To cope with blocking operations and unbounded memory requirements, some of these systems build on the well-established research on data streaming based on sliding windows and incremental operators [4], which includes systems such as Aurora [1] and Telegraph [11], often yielding approximate answers, rather than accurate results. Currently, among these DSPEs, the most popular platforms are Twitter's (now Apache) Storm [35], Spark's D-Streams [46], Flink Streaming [20], Apache S4 [42], and Apache Samza [39]. This paper addresses online processing for large-scale, incremental computations on a distributed processing platform. Our goal is to convert any batch data analysis program to an incremental distributed stream processing (DSP) program automatically, without requiring the user to modify this program. We are interested in deriving DSP programs that produce accurate results, rather than approximate answers. To accomplish this task, we would need to carefully analyze the program to identify those parts that can be used to process the incremental batches of data and those parts that can be used to merge the current results with the new results of processing the incremental batches. Such analysis is hard to attain for programs written in an algorithmic programming language but can become more tractable if it is performed on declarative queries. Fortunately, most programmers already prefer to use a higher-level query language, such as Apache Hive [26], to code their distributed data analysis applications, instead of coding them directly in an algorithmic language, such as Java. For instance, Hive is used for over 90% of Facebook Map-Reduce jobs. There are many reasons why programmers prefer query languages. First, it is hard to develop, optimize, and maintain nontrivial applications coded in a general-purpose programming language. Second, given the multitude of the new emerging distributed processing frameworks, such as Spark and Flink, it is hard to tell which one of them will prevail in the near future. Data intensive applications that have been coded in one of these paradigms may have to be rewritten as technologies evolve. Hence, it would be highly desirable to express these applications in a declarative query language that is independent of the underlying distributed platform. Furthermore, the execution of such queries can benefit from costbased query optimization and automatic parallelism, thus relieving the application developers from the intricacies of Big Data analytics and distributed computing. Therefore, our goal is to convert batch SQL-like queries to incremental DSP programs. We have developed our framework on Apache MRQL [33], because it is both platform-independent and powerful enough to express complex data analysis tasks, such as PageRank, data clustering, and matrix factorization, using SQL-like syntax exclusively. Since we are interested in deriving incremental programs that return accurate results, not approximate answers, our focus is in retaining a minimal state during the query evaluation lifetime and across iterations, and use incremental evaluation techniques to return an accurate snapshot answer at each time interval that depends on the current state and the latest batches of data.
We have developed general, sound methods to transform batch queries to incremental queries. The first step in our approach is to transform a query so that it propagates the join and group-by keys to the query output. This is known as lineage tracking ( [5,6,13]). That way, the values in the query output are grouped by a key combination, which corresponds the join/group-by keys used in deriving these values during query evaluation. If we also group the new data in the same way, then computations on current data can be combined with the computations on the new data by joining the data on these keys. This approach requires that we can combine computations on data that have the same lineage to derive incremental results. In our framework, this is accomplished by transforming a query to a 'monoid homomorphism' by extracting the non-homomorphic parts of the query outwards, using algebraic transformation rules, and combine them to form an answer function, which is detached from the rest of the query. We have implemented our incremental processing framework using Apache MRQL [33] on top of Apache Spark Streaming [46], which is an in-memory distributed stream processing platform. Our system is called Incremental MRQL. MRQL is currently the best choice for implementing our framework because other query languages for data-intensive, distributed computations provide limited syntax for operating on data collections, in the form of simple relational joins and group-bys, and cannot express complex data analysis tasks, such as PageRank, data clustering, and matrix factorization, using SQL-like syntax exclusively. Our framework though can be easily adapted to apply to other query languages, such as SQL, XQuery, Jaql, and Hive.
The contribution of this work can be summarized as fol-lows: • We present a general automated method to convert most distributed data analysis queries to incremental stream processing programs.
• Our methods can handle many forms of queries, including iterative and nested queries, group-by with aggregation, and joins on one-to-many relationships.
• We have extended our framework to handle deletions by using state increments to diminish the state in such a way that the query results would be the same as those we could have gotten on current data after deletion.
• We report on a prototype implementation of our framework using Apache MRQL running on top of Apache Spark Streaming. We show the effectiveness of our method through experiments on four queries: groupBy, join-groupBy, k-means clustering, and PageRank.
The rest of this paper is organized as follows. Section 2 illustrates our approach through examples. Section 3 compares our work with related work. Section 4 describes our earlier work on MRQL query processing. Section 5 defines the algebraic operators used in the MRQL algebra. Section 6 presents transformation rules for converting algebraic terms to a normal form that is easier to analyze. Section 7 describes an inference algorithm that statically infers the merge function that merges the current results with the results of processing the new data. Section 8 describes our method for transforming algebraic terms to propagate all keys used in joins and group-bys to the query output. Section 9 describes an algorithm that pulls the nonhomomorphic parts of a query outwards, deriving a homomorphism. Section 10, gives an example of the transformation of a batch query to an incremental query. Section 11 extends our framework to handle data deletion. Section 12 gives some implementation details. Section 13 presents experiments that evaluate the performance of our incremental query processing techniques for four queries.

HIGHLIGHTS OF OUR APPROACH
A dataset in our framework is a bag (multiset) that consists of arbitrarily complex values, which may contain nested bags and hierarchical data, such as XML and JSON fragments. We are considering continuous queries over a number of streaming data sources, Si, for 0 < i ≤ n. A data stream Si in our framework consists of an initial dataset, followed by a continuous stream of incremental batches ∆Si that arrive at regular time intervals ∆t. In addition to streaming data, there may be other input data sources that remain invariant through time. A streaming query in our framework can be expressed as q(S), where an Si ∈ S is a streaming data source. Incremental stream processing is feasible when we can derive the query results at time t+∆t by simply combining the query results at time t (i.e., the current results) with the results of processing the incremental batches ∆Si only, rather than the entire streams Si ⊎ ∆Si, where ⊎ is the additive (bag) union. This is possible if q(S ⊎ ∆S) can be expressed in terms of q(S) (the current query result) and q(∆S) (the incremental query result), that is, when q(S) is a homomorphism 1 over S. But some queries, such as counting 1 We use the term homomorphism throughout the paper as an abbreviation of monoid homomorphism. For such queries, we break q into two functions a and h, so that q(S) = a(h(S)) and h is a homomorphism. The function h is a homomorphism if h(S ⊎ ∆S) = h(S) ⊗ h(∆S) for some monoid ⊗ (an associative function with a zero element ⊗z). For example, the query that counts the number of distinct elements can be broken into the query h that returns the list of distinct elements (a homomorphism), followed by the answer query a that counts these elements. Ideally, we would like most of the computation in q to be done in h, leaving only some computationally inexpensive data mappings to the answer function a. Note that, the obvious solution when a is equal to q and h is the union of data sources, is also the worst-case scenario that we try to avoid, since it basically requires to compute the new result from the entire input, S ⊎ ∆S. On the other hand, in the special case when a is the identity function and ⊗ is equal to ⊎, the output at each time interval can be simply taken to be only h(∆S), which is the output we would expect to get from a fixed window system (i.e., new batches of output from new batches of input).
If we split q into a homomorphism h and an answer function a, then we can calculate h incrementally by storing its results into a state and then using the current state to calculate the next h result. Initially, state = ⊗z or, if there are initial stream data, state = h(S). Then, at each time interval ∆t, the query answer is calculated from the state, which becomes equal to h(S ⊎ ∆S): In Spark, for example, the state and the invariant data sources are stored in memory as Distributed DataSets (Spark's RDDs [45]) and are distributed across the worker nodes. However, the streaming data sources are implemented as Discretized Streams (Spark's D-Streams [46]), which are also distributed.
Our framework works best for queries whose output is considerably smaller than their input, such as for data analysis queries that aggregate data. Such queries would require a smaller state and impose less processing overhead to ⊗. Figure 1 shows the evaluation of an incremental query q(S1, S2) = a(h(S1, S2)) over two streaming data sources, where h(S1 ⊎ ∆S1, S2 ⊎ ∆S2) = h(S1, S2) ⊗ h(∆S1, ∆S2). Our query processing system performs the following tasks: 1. It pulls all non-homomorphic parts of a query q out from the query, using algebraic transformations.
2. It collects these non-homomorphic parts into an answer function a, leaving an algebraic homomorphic term h, such that q(S) = a(h(S)).
These tasks are, in general, hard to attain for a program expressed in an algorithmic programming language. Instead, these tasks can become more tractable if they are performed on higher-order operations, such as the MRQL query algebra [18,17]. Although all algebraic operations used in MRQL are homomorphic, their composition may not be.
We have developed transformation rules to derive homomorphisms from compositions of homomorphisms, and for pulling non-homomorphic parts outside a query. Our methods can handle most forms of queries on nested data sets, including iterative queries, complex nested queries with any form and any number of nesting levels, general group-bys with aggregations, and general one-to-one and one-to-many equi-joins. Our methods cannot handle non-equi-joins and many-to-many equi-joins, as they are notoriously difficult to implement efficiently in a streaming or an incremental computing environment.
Example. For example, consider the following query (all queries are expressed in MRQL): where S1 and S2 in q(S1, S2) are streaming data sources. Unfortunately, q(S1, S2) is not a homomorphism over S1 and S2, that is, q(S1 ⊎ ∆S1, S2 ⊎ ∆S2) cannot be expressed in terms of q(S1, S2) and q(∆S1, ∆S2) exclusively. The intrinsic reason behind this is that there is no lineage in the query output that links a pair in the query result to the join key (x.B or y.C) that contributed to this pair. Consequently, there is no way to tell how the new data batches ∆S1 and ∆S2 will contribute to the previous query results if we do not know how these results are related to the inputs S1 and S2. To compensate, we need to establish links between the query results and the data sources that were used to form their values. This is called lineage tracking and has been used for consistent representation of uncertain data [5] and for propagating annotations in relational queries [6]. In our case, this lineage tracking can be accomplished by propagating all keys used in joins and group-bys along with the values associated with the keys, so that, for each combination of keys, we have one group of result values. For our query, this is done by including the join key x.B as a group-by key. That is, in the following transformed query: h(S1, S2) = select ((x.A, x.B), (sum(y.D), count(y.D))) from x in S1 , y in S2 where x.B = y.C group by x.A, x.B the join key is propagated to the output values so that the avg components, sum and count, are aggregations over groups that correspond to unique combinations of x.A and x.B. It can be shown that this query is a homomorphism over S1 and S2, provided that the join is not on a many-tomany relationship. In general, a query with N joins/groupbys/order-bys will be transformed to a query that injects the join/group-by/order-by keys to the output so that each output value is annotated with a combination of N keys. Note that the output of the query h is larger than that of the original query q because it creates more groups and each group is assigned two values (sum and count), instead of one (avg). This is expected since we extended q with lineage tracking. The answer query a that gives the final result in q(S1, S2) = a(h(S1, S2)) is: a(X) = select (k, sum(s)/sum(c)) from ((k, j ),( s ,c)) in X group by k that is, it removes the lineage j from X but also groups the result by the group-by key again and calculates the final avg values. The merge function for the homomorphism h is a full outer join on the lineage key θ that aggregates the matches. It is specified as: where θ matches the lineage (k, j) and π1 is pair and bag projection.
Merging States. The effectiveness of our incremental processing depends on the efficient implementation of the state transformation that merges the previous state with the results of processing the new data, state ⊗ h(∆S). The merge operation X ⊗ Y in the previous example can be implemented efficiently as a partitioned join. On Spark, for example, both the state and the new results are kept in the distributed memory of the worker nodes (as RDDs), while the full outer join can be implemented as a coGroup operation, which shuffles the join input data across the worker nodes using hash partitioning. However, when the new state is created by coGroup, it is already partitioned by the join key and is ready to be used for the next call to coGroup to handle the next batch of data. Consequently, only the results of processing the new data, which are typically smaller than the state, would have to be shuffled across the worker nodes before coGroup. Although other queries may require different merge functions, the correlation between the previous state and the results of processing the new data is always based on the lineage keys. Therefore, regardless of the query, we can keep the state partitioned on the lineage keys by simply leaving the new state partitions at the place they were generated. However, the new results would have to be partitioned and shuffled across the working nodes to be combined with the current state. Our approach is based on the assumption that, since the state is kept in the distributed memory, there will be very little overhead in replacing the current state with a new state, as long as it is not repartitioned. But this assumption may not be valid if the input data or the current state is larger than the available distributed memory. Indeed, one of our goals is to be able to process data larger than the available memory, by processing these data incrementally, in batches that can fit in memory. However, if the state is stored on disk, replacing it with a new state may become prohibitively expensive. Fortunately, there is no need for using a distributed key-value store, such as HBase, to update the state in-place. There is, in fact, a simpler solution: since the state is kept partitioned on the lineage keys, every worker node may store its assigned state partition on its local disk, as a binary file (such as, an HDFS Sequence file) sorted by the lineage key. Then, after the results of processing the new data are distributed to worker nodes (using uniform hashing based on lineage key), each worker node will sort its new data partition by the lineage key and will merge it with its old state (stored on its local disk) to create a new state.
Iteration. Given that many important data analysis and mining algorithms, such as PageRank and k-means clustering, require repetition, we have extended our methods to include repetition, so that these algorithms too can become incremental. A repetition can take the following general form: where X is the fixpoint of the repetition that has initial value g(S). That is, X = f n (g(S), S), where f n applies f n times.
Note that X is not necessarily a bag. For example, nonnegative Matrix Factorization [22], used in machine learning applications, such as for recommender systems, splits a matrix S into two non-negative matrices W and H. In that case, the fixpoint X is the pair (W, H), which is refined at every loop step. An exact incremental solution of the above repetition is only possible if f is a homomorphism; a strict requirement that excludes many important iterative algorithms, such as PageRank and k-means clustering. Instead, our approach is to use an approximate solution that works well for iterative queries that improve a solution at each iteration step. As before, we split f into a homomorphism h, for some monoid ⊗, and an answer function a, so that f (X, S) = a(h(X, S)) and h(X, S ⊎ ∆S) = h(X, S) ⊗ h(X, ∆S). Let T be the current state on input S and T ′ be the new state on input S ⊎ ∆S. An ideal solution would have been to derive the new state T ′ incrementally, as T ⊗∆T , so that the state increment ∆T depends on ∆S but not on T . But this independence from T is not always possible for many iterative algorithms. In PageRank, for example, we cannot just calculate the PageRanks of the new data and merge them with the PageRank of the existing data because the new PageRank contributions may have to propagate to the rest of the graph. On the other hand, it would be too expensive to correlate the entire state T with ∆T at each iteration step. Our compromise is to partially correlate T with ∆T at each iteration step, and then fully merge ∆T with T after the iteration. This partial correlation is accomplished with the operation ⊗ , called diffusion, which is directly derived from the merge function ⊗. That is, while T ⊗ ∆T returns a new complete state by merging T with ∆T , the operation T ⊗ ∆T returns a new ∆T that contains only the part of T ⊗ ∆T that is different from T . Although T ⊗ ∆T is larger than ∆T , it is often far smaller than T ⊗∆T . For instance, if T were a bag, T ⊗ ∆T would have been a subset of T ⊗ ∆T . Based on this analysis, we are using the following approximate algorithm to compute the iteration incrementally: The diffusion operator ⊗ must satisfy the property: that is, when T ⊗ ∆T and T ⊗ ∆T are merged with T , they give the same answer, because T ⊗ ∆T discards the parts of T that are not joined with ∆T , but these parts are embedded back in the answer when we merge with T .
For example, PageRank is an iterative algorithm that calculates the PageRank of a graph node as the sum of the incoming PageRank contributions from its neighbors, while its own PageRank is equally distributed to its outgoing neighbors. The PageRank query expressed in MRQL is a simple self-join on the graph, which is optimized into a single group-by operation [18]. For PageRank, the state merging ⊗ is a full outer join that incorporates the new PageRank contributions to the existing PageRanks. The diffusion operation T ⊗ ∆T , on the other hand, propagates the new PageRank contributions from ∆T to T , that is, from the nodes in ∆T to their immediate outgoing neighbors in both T and ∆T . Thus, at each iteration step, ∆T is expanded to T ⊗ ∆T , growing one level at a time, in a way similar to breadth-first-search. Consequently, our approximate algorithm propagates PageRanks up to depth n, starting from the ∆T nodes, and only the affected nodes will be part of the new ∆T . The ⊗ operation is similar to ⊗, but with a right-outer join instead of a full outer join. That way, the data from T that are not joined with ∆T will not appear in the new ∆T .
A more complete example is the following query that implements the k-means clustering algorithm by repeatedly deriving k new centroids from the old: where S1 is the input stream of points on the X-Y plane, centroids is the current set of centroids (k cluster centers), and distance is a function that calculates the distance between two points. The initial value of centroids (the ... value) can be a bag of k random points. The inner select-query in the group-by part assigns the closest centroid to a point s (where [0] returns the first tuple of an ordered list). The outer select-query in the repeat step clusters the data points by their closest centroid, and, for each cluster, a new centroid is calculated from the average values of its points. As in the previous join-groupBy example, the average value of a bag of values is decomposed into a pair that contains the sum and the count of values. That is, the state is a bag of {(k, ((sx, cx), (sy, cy)))} so that the centroids are the bag of points {(sx/cx, sy/cy)} and k is the lineage (the group-by key), which is the X-Y coordinates of a centroid. Consequently, the answer query that returns the final result (the centroids) is: a( state ) = select < X: sx/cx, Y: sy/cy > from (k,(( sx ,cx ),( sy ,cy))) in state (it does not require a group-by since k is the only lineage key), while h(X, S1) is: The merge function ⊗ is a full outer join, similar to the one used by the join-groupBy example. The diffusion operation X ⊗ Y though is a right-outer join that discards those state data that do not join with the new data. That is, it is equal to the X ⊗ Y query without the last union: But, suppose that one of the group-by or join keys in a query is a floating point number. This is the case with the previous k-means clustering query, because it groups the points by their closest centroids, which contain floating point numbers. The group-by operation itself is not a problem because the centroids are fixed during the group-by. The problem arises when merging the current state with the results of processing the new data. For the k-means query example, the merge function ⊗ is an equi-join whose join attribute is a centroid, so that the sum and count values associated with the same centroid are brought together from the join inputs and are accumulated. Since the join condition is over attributes with floating point numbers, the join condition will fail in most cases. This problem becomes even worse for the approximate solution, because it uses different sets of centroids X and Xprev when states are merged. Most iterative queries do not have this problem. The lineage in PageRank, for example, is the node ID, which remains invariant across iterations. For these uncommon queries, such as k-means, that have floating point numbers in their join/group-by/order-by attributes, we use yet another approximation: the join is done based on an "approximate equality" where two floating point numbers are taken to be equal if their difference is below some given threshold. This works well for our approximate solution for iteration because it is based on the assumption that the new solution X is approximately equal to the previous one, Xprev.

RELATED WORK
New frameworks in distributed Big Data analytics have become essential tools to large-scale machine learning and scientific discoveries. Among these frameworks, the Map-Reduce programming model [14] has emerged as a generic, scalable, and cost effective solution for Big Data processing on clusters of commodity hardware. One of the major drawbacks of the Map-Reduce model is that, to simplify reliability and fault tolerance, it does not preserve data in memory between the map and reduce tasks of a Map-Reduce job or across consecutive jobs, which imposes a high overhead to complex workflows and graph algorithms, such as PageRank, which require repetitive Map-Reduce jobs. Recent systems for cloud computing use distributed memory for inter-node communication, such as the main memory Map-Reduce (M3R [38]), Apache Spark [40], Apache Flink [20], Piccolo [36], and distributed GraphLab [28]. Another alternative framework to the Map-Reduce model is the Bulk Synchronous Parallelism (BSP) programming model [44]. The best known implementations of the BSP model for data analysis on the cloud are Google's Pregel [29], Apache Giraph [21], and Apache Hama [25].
Although the Map-Reduce framework was originally designed for batch processing, there are several recent systems that have extended Map-Reduce with online processing capabilities. Some of these systems build on the wellestablished research on data streaming based on sliding windows and incremental operators [4], which includes systems such as Aurora [1] and Telegraph [11]. MapReduce Online [12] maintains state in memory for a chain of MapReduce jobs and reacts efficiently to additional input records. It also provides a memoization-aware scheduler to reduce communication across a cluster. Incoop [7] is a Hadoopbased incremental processing system with an incremental storage system that identifies the similarities between the input data of consecutive job runs and splits the input based on the similarity and file content. i 2 MapReduce [47] implements incremental iterative Map-Reduce jobs using a store, MRB-Store, that maps input values to the reduce output values. This store is used for detecting delta changes and propagating these changes to the output. Google's Percolator [35] is a system based on BigTable for incrementally processing updates to a large data set. It updates an index incrementally as new documents are crawled. Microsoft Naiad [32] is a distributed framework for cyclic dataflow programs that facilitates iterative and incremental computations. It is based on differential dataflow computations, which allow incremental computations to have nested iterations. CBP [27] is a continuous bulk processing system on Hadoop that provides a stateful group-wise operator that allows users to easily store and retrieve state during the reduce stage as new data inputs arrive. Their incremental computing PageRank implementation is able to cut running time in half. REX [31] handles iterative computations in which changes in the form of deltas are propagated across iterations and state is updated efficiently. In contrast to our automated approach, REX requires the programmer to explicitly specify how to process deltas, which are handled as first class objects. Trill [10] is a high throughput, low latency streaming query processor for temporal relational data, developed at Microsoft Research. The Reactive Aggregator [43], developed at IBM Research, is a new sliding-window streaming engine that performs many forms of sliding-window aggregation incrementally. In addition to these general data analysis engines, there are many data analysis algorithms that have been implemented incrementally, such as incremental pagerank [15]. Finally, the incremental query processing is related to the problem of incremental view maintenance, which has been extensively studied in the context of relational views (see [23] for a literature survey).
Many novel Big Data stream processing systems, also known as distributed stream processing engines (DSPEs), have emerged recently. The most popular one is Twitter's Storm [35], which is now part of the Apache ecosystem for Big Data analytics. It provides primitives for transforming streams based on a user-defined topology, consisting of spouts (stream sources) and bolts (which consume input streams and may emit new streams). Other popular DSPE platforms include Spark's D-Streams [46], Flink Streaming [20], Apache S4 [42], and Apache Samza [39].
In programming languages, self-adjusting computation [2] refers to a technique for compiling batch programs into programs that can automatically respond to changes to their data. It requires the construction of a dependence graph at run-time so that when the computation data changes, the output can be updated by re-evaluating only the affected parts of the computation. In contrast to our work, which requires only the state to reside in memory, self-adjusting computation expects both the input and the output of a computation to reside in memory, which makes it inappropriate for unbounded data in a continuous stream. Furthermore, such dynamic methods impose a run-time storage and computation overhead by maintaining the dependence graph. The main idea in [2] is to manually annotate the parts of the input type that is changeable, and the system will derive an incremental program automatically based on these annotations. Each changeable value is wrapped by a mutator that includes a list of reader closures that need to be evaluated when the value changes. A read operation on a mutator inserts a new closure, while the write operation triggers the evaluation of the closures, which may cause writes to other mutators, etc, resulting to a cascade of closure execution triggered by changed data only. This technique has been extended to handle incremental list insertions (like our work), but it requires the rewriting of all list operations to work on incremental lists. Recently, there is a proof-of-concept implementation of this technique on map-reduce [3], but it was tested on a serial machine. It is doubtful that such dynamic techniques can be efficiently applied to a distributed environment, where a write in one compute node may cause a read in another node. Finally, there is recent work on static incrementalization based on derivatives [8]. In contrast to our work, it assumes that the merge function that combines the previous result with the result on the delta changes uses exactly the same delta changes, a restriction that excludes aggregations and groupbys.

EARLIER WORK: MRQL
Apache MRQL [33] is a query processing and optimization system for large-scale, distributed data analysis. MRQL was originally developed by the author ( [18,17]), but is now an Apache incubating project with many developers and users worldwide. The MRQL language is an SQL-like query language for large-scale data analysis on computer clusters. The MRQL query processing system can evaluate MRQL queries in four modes: in Map-Reduce mode using Apache Hadoop [24], in BSP mode (Bulk Synchronous Parallel model) using Apache Hama [25], in Spark mode using Apache Spark [40], and in Flink mode using Apache Flink [20]. The MRQL query language is powerful enough to express most common data analysis tasks over many forms of raw in-situ data, such as XML and JSON documents, binary files, and CSV documents. The design of MRQL has been influenced by XQuery and ODMG OQL, although it uses SQL-like syntax. In fact, when restricted to XML, MRQL is as powerful as XQuery. MRQL is more powerful than other current high-level Map-Reduce languages, such as Hive [26] and PigLatin [34], since it can operate on more complex data and supports more powerful query constructs, thus eliminating the need for using explicit procedural code. With MRQL, users are able to express complex data analysis tasks, such as PageRank, k-means clustering, matrix factorization, etc, using SQL-like queries exclusively, while the MRQL query processing system is able to compile these queries to efficient Java code that can run on various distributed processing platforms. For example, the PageRank query on raw DBLP XML data, which ranks authors based on the number of citations they have received from other authors, is 16 lines long [17] and can be executed on all the supported platforms as is, without changing the query.
A recent extension to MRQL, called MRQL Streaming, supports the processing of continuous MRQL queries over streams of batch data (that is, data that come in continuous large batches). Before the incremental MRQL work presented in this paper, MRQL Streaming supported traditional window-based streaming based on a fixed window during a specified time interval. Any batch MRQL query can be converted to a window-based streaming query by replacing at least one of the 'source' calls in the query that access data sources to 'stream' calls (with exactly the same call arguments). For example, the query: select (k,avg(p.Y)) from p in stream(binary ,"points") group by k: p.X groups a stream of points by their X coordinate and returns the average Y values in each group. The MRQL Streaming engine first processes all the existing sequence files in the directory points and then checks this directory periodically for new files. When new files are inserted in the directory, it processes the new batch of data using distributed query processing. MRQL Streaming also supports a stream input format for listening to TCP sockets for text input based on one of the MRQL Parsed Input Formats (XML, JSON, CSV). A query may work on multiple stream sources and multiple batch sources. If there is at least one stream source, the query becomes continuous (it never stops). The output of a continuous query is stored in a file directory, where each file contains the results of processing each batch of streaming data. Currently, MRQL Streaming works on Spark Streaming only but there are current efforts to add support for Storm and Flink Streaming in the near future. The work reported here, called Incremental MRQL, extends the current MRQL Streaming engine with incremental stream processing.

THE MRQL ALGEBRA
Our compiler translates queries to algebraic terms and then uses rewrite rules to put these algebraic terms into a homomorphic form, which is then used to compute the query results incrementally by combining the previous results with the results of processing the incremental batches.
The MRQL algebra described in this section is a variation of the algebra presented in our previous work [18], but is more suitable for describing our incremental methods. The relational algebra, the nested relational algebra, as well as many other database algebras can be easily translated to our algebra. Our algebra consists of a small number of higherorder homomorphic operators [18], which are defined using structural recursion based on the union representation of bags [19]. Monoid homomorphisms capture the essence of many divide-and-conquer algorithms and can be used as the basis for data parallelism [19].
The first operator, cMap (also known as concat-map or flatten-map in functional programming languages), generalizes the select, project, join, and unnest operators of the nested relational algebra. Given two arbitrary types α and β, the operation cMap(f, X) maps a bag X of type {α} to a bag of type {β} by applying the function f of type α → {β} to each element of X, yielding one bag for each element, and then by merging these bags to form a single bag of type {β}. Using a set former notation on bags, it is expressed as: or, alternatively, using structural recursion: Given an arbitrary type κ that supports value equality (=), an arbitrary type α, and a bag The groupBy operation cannot be defined using a set former notation, but can be defined using structural recursion: where the parametric monoid ⇑⊕ is a full outer join that merges groups associated with the same key using the monoid ⊕ (equal to ⊎ for groupBy): In other words, the monoid ⇑⊎ constructs a set of pairs whose unique key is the first pair element. In fact, any bag X can be converted to a set using π1(groupBy(cMap(λx. {(x, x)}, X))). Note also that groupBy(X) is not the same as the nesting { (k, { y | (k ′ , y) ∈ X, k = k ′ }) | (k, x) ∈ X }, as the latter contains duplicate entries for the key k. Unlike nesting, unnesting a groupBy returns the input bag (proven in Appendix A): Although any join X ⊲⊳p Y can be expressed as a nested cMap: this term is not always a homomorphism on both inputs. Instead, MRQL provides a special homomorphic operation for equi-joins and outer joins, coGroup(X, Y ), between a bag X of type {(κ, α)} and a bag Y of type {(κ, β)} over their first component of a type κ, which returns a bag of type {(κ, ({α}, {β}))}: where the product of two monoids, ⊕ × ⊗ is a monoid that, when applied to two pairs (x1, x2) and (y1, y2), returns (x1 ⊕y1, x2 ⊗y2).  (3)) that both coGroup inputs can be derived from the coGroup result: and a coGroup is equivalent to a groupBy if one of the inputs is empty: Aggregations are captured by the operation reduce(⊕, X), which aggregates a bag X using a commutative monoid ⊕. For example, reduce(+, {1, 2, 3}) = 6. This operation is in fact a general homomorphism that can be defined on any monoid ⊕ with an identity function. We have: reduce(⊎, X) = X reduce(⇑⊎, X) = groupBy(X) Finally, iteration repeat(F, n, X) over the bag X of type {α} applies F of type {α} → {α} to X n times, yielding a bag of type {α}: repeat(F, n, X) = F n (X) = F (F (. . . F (X))) An iteration is a homomorphism as long as F is a homomorphism, that is, when F (X ⊎ Y ) = F (X) ⊎ F (Y ).
Definition 1 (MRQL Algebra). The MRQL algebra consists of terms that take the following form: where ⊕ is a monoid on basic types, such as +, * , or, and, etc. Function f is an anonymous function that may contain such algebraic terms but is not permitted to contain any reference to a stream source, Si.
The restriction on f in Definition 1 excludes non-equi-joins, such as cross products, which require a nested cMap in which the inner cMap is over a data source. This algebra does not include iteration, repeat(F, n, e). Iterations have been discussed in Section 2. In addition, for brevity, this algebra does not include terms for non-streaming input sources, general tuple and record construction and projection, bag union, singleton and empty bag, arithmetic operations, ifthen-else expressions, boolean operations, etc. In addition to these operations, there are a few more algebraic operations that can be expressed as homomorphisms, such as orderBy(X), which is a groupBy followed by a sorting over the group-by key. This operation returns a list, which is represented by the non-commutative monoid listappend, + +. Mixing multiple collection monoids and operations in the same algebra has been addressed by our previous work [19], and is left out from this paper to keep our analysis simple.

QUERY NORMALIZATION
In an earlier work [18], we have presented general algorithms for unnesting nested queries. For example, consider the following nested query: A typical method for evaluating this query in a relational system is to first group Y by y.B, yielding pairs of y.B and sum(y.C), and then to join the result with X on x.A=y.B using a left-outer join, removing all those matches whose x.D is below the sum. But, in our framework, this query is translated into: cMap( λ(k,(xs,ys)). cMap( λx. if x.D > reduce(+,ys) cMap( λy. {(y.B,y.C)}, Y ) ) ) That is, the query unnesting is done with a left-outer join, which is captured concisely by the coGroup operation without the need for using an additional group-by operation or handling null values. This unnesting technique was generalized to handle arbitrary nested queries, at any place, number, and nesting level (the reader is referred to our earlier work [18] for details).
The algebraic terms derived from MRQL queries can be normalized using the following rule: cMap(f, cMap(g, S)) → cMap(λx. cMap(f, g(x)), S) (4) which fuses two cascaded cMaps into a nested cMap, thus avoiding the construction of the intermediate bag. This rule can be proven directly from the cMap definition in Equation (1): If we apply the transformation (4) repeatedly, and given that we can always use the identity cMap(λx.{x}, X) = X in places where there is no cMap between groupBy/coGroup operations, any algebraic terms in Definition 1 can be normalized into the following form: Definition 2 (Normalized MRQL Algebra). The normalized MRQL algebra consists of terms q that take the following form: where function f is an anonymous function that does not contain any reference to a stream source, Si.
The query body is a tree of groupBy/coGroup operations connected via cMaps.

MONOID INFERENCE
One of our tasks is, given an algebraic term f (S), where an Si ∈ S is a streaming data source, to prove that f is a homomorphism by deriving a monoid ⊗ such that: f (S1 ⊎S ′ 1 , . . . , Sn ⊎S ′ n ) = f (S1, . . . , Sn)⊗f (S ′ 1 , . . . , S ′ n ) (6) We have developed a monoid inference system, inspired by type inference systems used in programming languages. We use the judgment ρ ⊢ e : ⊕ to indicate that e is a monoid homomorphism with a merge function ⊕ under the environment ρ, which binds variables to monoids. The notation ρ(v) extracts the binding of the variable v, while ρ[v : ⊕] extends the environment with a new binding from v to ⊕. Equation (6) can now be expressed as the judgment: [S1 : ⊎, . . . , Sn : ⊎] ⊢ f (S) : ⊗ If a term is invariant under change, such as an invariant data source, it is associated with the special monoid ✷: Our monoid inference algorithm is a heuristic algorithm that annotates terms with monoids (when possible). It is very similar to type inference. Most of our inference rules are expressed as fractions: the denominator (below the line) contains the premises (separated by comma) and the numerator (above the line) is the conclusion. For example, the rule ρ ⊢f (x): ⊗ ρ ⊢x: ⊕ indicates that f (x1 ⊕x2) = f (x1)⊗f (x2). Figure 2 gives some of the inference rules. More rules will be given in Lemma 1. Rules (5c) through (5f) are derived directly from the algebraic definition of the operators. Rule (5a) retrieves the associated monoid of a variable v from the environment ρ. Rule (5b) indicates that if all the variables in a term e are invariant, then so is e. Rule (5h) indicates that a cMap over a groupBy is a homomorphism as long as its functional argument is a homomorphism. It can be proven as follows: ⊎ · · · (given that k and θ are unique keys) = cMap(f, X) ⇑⊗ cMap(f, Y )

INJECTING LINEAGE TRACKING
There are two tasks that need to be accomplished to achieve our goal of transforming an algebraic term into a homomorphism: 1) transform the algebraic term in such a way that it propagates all keys used in joins and group-bys to the query output, and 2) pull the non-homomorphic parts out of the algebraic term so that it becomes a homomorphism. In this section, we address the first task.
We will transform the algebraic terms given in Definition 2 in such a way that they propagate the join and the groupby keys. That is, each value v returned by these terms is annotated with a lineage θ, as a pair (θ, v), where θ takes the following form: θ, θ1, θ2 ::= (κ, θ) extended with groupBy key κ | (κ, (θ1, θ2)) extended with coGroup key κ | () empty lineage That is, the lineage θ of the query result v is the tree of the groupBy and coGroup keys that are used in deriving the result v (one key for each groupBy and coGroup operation). The lineage tree θ has the same shape as the group-By/coGroup tree of the query. We transform a query q in such a way that, if the output of the query is {t} for some type t, then the transformed query will have output {(θ, t)}. Furthermore, if the the output is a non-collection type t, then the transformed query will also have output {(θ, t)}, which separates the contributions to t associated with each combination of group-by/join keys.

Algorithm 1 (Lineage Annotation).
Input: a normalized query term q defined in Definition 2 Output: a term Tq[[q]] annotated with a lineage θ  where sMap1, sMap2, sMap3, swap, and mix are defined as follows: ((k, θ), b)) | ((k, θ), a) ∈ X, mix(X) { ((k, (θx, θy)), (xs, ys)) | (k, (s1, s2)) ∈ X, (8e) (θx, xs) ∈ groupBy(s1), (θy, ys) ∈ groupBy(s2) } A query q in our framework is transformed in such a way that it propagates the lineage from the data stream sources to the query output, starting with the empty lineage () at the sources and extended with the join and group-by keys. The sMap1 operation is a cMap that propagates the input lineage θ to the output as is. The sMap2 operation is a cMap that extends the input lineage θ with a groupBy/coGroup key k. The lineage propagation is done by the cMap Rules (7c) and (7h). Rule (7c) applies to the outer query cMap that produces the query output. It simply propagates the lineage from the cMap input to the output. Rule (7h) applies to a cMap that returns the input of a groupBy or coGroup. The output of this cMap must be a bag of key-value pairs, as is expected by a groupBy or a coGroup. Thus, Rule (7h) extends the lineage with a new key and prepares the cMap output for the enclosing groupBy or coGroup. This is done by translating cMap to sMap2. Rule (7b) indicates that a total aggregation becomes a group-by aggregation by aggregating the values of each group associated with a different lineage θ. Rule (7e) translates a groupBy on a key to a groupBy on the entire lineage (which includes the groupBy key). Rule (7f) translates a coGroup on a key to a coGroup on the entire lineage, but it is done using the function mix (defined in (8e) because the left input lineage θx is not necessarily compatible with the right input lineage θy. Given this, Rule (7f) generates a coGroup on the join key first, and then, for each join key k, it groups the left and right join matches by θx and θy respectively, so that the output contains unique lineage key combinations, (k, (θx, θy)). Finally, Rule (7g) annotates each value of the input stream Si with the empty lineage ().
We will prove next that the transformed query Tq[[q]] is a homomorphism. We first prove that the generated sMap1 and sMap2 in Tq [[q]] are homomorphisms, provided that their functional arguments are homomorphisms. More specifically, we prove the following judgments: The proof of this Lemma is given in Appendix A. Based on Lemma 1, we can now prove that the transformed query is a homomorphism: The proof is given in Appendix A.

Definition 4 (Query Answer). The query answer A[[q]]
x of the query q, defined in Definition 2, is a function over the current state x = Tq[[q]] and is derived as follows: The proof of this theorem is given in Appendix A.

Restricting Joins
Theorem 1 indicates that a query transformed by Algorithm 1 is a homomorphism if the cMap functional arguments in the query satisfy the premises in Judgment (9a) and (9b). But, consider the following join: cMap(λ(k, (xs, ys)). cMap(λx. cMap(λy. {(x, y)}, xs), ys), coGroup(X, Y )) Our monoid inference system can not prove that the functional argument of this cMap is a homomorphism on both xs and ys. This is expected because this join could be a manyto-many join, which we know can not be a homomorphism.
Since we have given up on handling many-to-many joins, we want to extend the monoid inference algorithm to always assume that every join is a one-to-one or one-to-many join and draw inferences based on this assumption. We have already seen in Section 7 that if a term is invariant under change, such as an invariant data source, it is associated with the special monoid ✷. We can also use the annotation ✷ to denote certain functional dependencies, such as κ → α on a bag X of type {(κ, α)}, which indicates that the second component of a pair in X depends on the first. This dependency is captured by annotating groupBy(X) with the monoid ⇑✷, which indicates that each group remains invariant under change, implying that the group-by key is also a unique key of X. That is, if (k, s1) ∈ X1 and (k, s2) ∈ X2, then { (k, s1✷s2) | (k, s1) ∈ X1, (k, s2) ∈ X2 } in Definition (2) must have s1 = s2 so that s1✷s2 = s1, otherwise it will be an error. This means that we can not have two different groups associated with the same key k. Given that a groupBy over a singleton gives a singleton group, each group returned from a groupBy is a singleton that remains invariant. A similar functional dependency can also apply to a join between X of type {(κ, α)} and Y of type {(κ, β)}, which is a coGroup(X, Y ) of type { (κ, ({α}, {β}))}. To indicate that this join is one-to-one or one-to-many, we annotate coGroup(X, Y ) with the monoid ⇑✷×⊎, which enforces the constraint that the bag {α} in {(κ, ({α}, {β}))} be either empty or singleton, that is, at most one X element can be joined with Y over a key κ. This is because the X and Y values that are joined over the same key are merged with ✷ and ⊎, respectively, as indicated by ⇑✷×⊎. Therefore, to incorporate the assumption that all joins are one-to-one or one-to-many in the monoid inference system, we have to replace Judgment (5g) with the following judgment: Given this judgment, the cMap input of our join example will be annotated with ⇑✷×⊎, which means that xs and ys will be annotated with ✷ and ⊎, respectively (ie, xs is invariant). Based on these annotations, the functional argument cMap(λx. cMap(λy. {(x, y)}, xs), ys) satisfies the premise in Judgment (9b) since it is annotated with ⊎.

Handling Iterative Queries
The approximate algorithm that processes iterations incrementally, presented in Section 2, requires an additional operation, called the diffusion operator ⊗ , that satisfies the property: Our goal is to define ⊗ in terms of ⊗ in such a way that T ⊗ ∆T contains only those parts of T ⊗ ∆T that changed from T .
Definition 5 (Diffusion). The diffusion ⊗ of a monoid ⊗ is defined as follows: for any other monoid where ⇓⊗ is a right-outer join defined as follows: This ⊗ satisfies the desired property, T ⊗ (T ⊗ ∆T ) = T ⊗ (T ⊗ ∆T ) (proven in Appendix A).

NON-HOMOMORPHIC TERMS
Theorem 2 indicates that the terms generated by the transformations (7a) through (7h) are homomorphisms as long as the premises of the judgements in Lemma 1 are true. These premises indicate that the cMap functional arguments must be homomorphisms too. We want the operations that cannot be annotated with a monoid to be transformed so that the non-homomorphic parts of the operation are pulled outwards from the query using rewrite rules. We achieve this with the help of kMap: which is a cMap that propagates the lineage θ as is. In our framework, all non-homomorphic parts take the form of a kMap and are accumulated into one kMap using kMap(f, kMap(g, X)) → kMap(λx. f (g(x)), X) More specifically, we first split each non-homomorphic cMap to a composition of a kMap and a cMap so that the latter cMap is a homomorphism, and then we pull and merge kMaps. Consider the term cMap(λ(θ, v). {(θ ′ , e)}, X), which creates a new lineage θ ′ from the old θ. In our framework, we find the largest subterms in the algebraic term e, namely e1, . . . , en, that are homomorphisms. This is accomplished by traversing the tree that represents the term e, starting from the root, and by checking if the node can be inferred to be a homomorphism. If it is, the node is replaced with a new variable. Thus, e is mapped to a term f (e1, . . . , en), for some term f , and the terms e1, . . . , en are replaced with variables when f is pulled outwards: The kMaps are combined and are pulled outwards from the query using the following rewrite rules: sMap2(g, kMap(f, X)) → sMap2(λ(k, v). g(k, f (v)), X) sMap1(g, kMap(f, X)) → sMap1(λ(k, v). g(k, f (v)), X) groupBy(kMap(f, X)) → kMap(λs. map(f, s), groupBy(X)) coGroup(kMap(fx, X), kMap(fy, Y )) → kMap(λ(k, (xs, ys)). (k, (map(fx, xs), map(fy, ys))), coGroup(X, Y )) where map(f, X) = cMap(λx. {f (x)}, X). Rewrite rules as these, when applied repeatedly, can pull out and combine the non-homomorphic parts of a query, leaving a homomorphism whose merge function can be derived from our annotation rules.

HANDLING DELETIONS
Our framework can be easily extended to handle deletion of existing data from an input stream, in addition to insertion of new data. To handle both insertions and deletions, a data stream Si in our framework consists of an initial data set, followed by a continuous stream of incremental batches ∆Si and a continuous stream of decremental batches ∆S ′ i that arrive at regular time intervals ∆t. Then, the stream data at time t + ∆t is (An element appears in X − Y as many times as it appears in X, minus the number of times it appears in Y .) This dual stream of updates, can be implemented as a single stream that contains updates tagged with + or -, to indicate insertion or deletion. Alternatively, a stream data source may monitor two separate directories, one for insertions and another for deletions, so that if a new file is created, it will be treated as a new batch of insertions or deletions, depending on the directory. In our framework, we require that ∆S ′ i ⊆ Si, that is, we can only delete values that have already appeared in the stream in larger or equal multiplicities. This restriction implies that there is a bag Wi such that Si = Wi ⊎ ∆S ′ i . Without this restriction, it would be hard to diminish the query results on Si by the results on ∆Si to calculate the new results.
Definition 6 (Diminisher). The diminisher ⊗ of a monoid ⊗ is defined as follows: where ⇓⊗ is a left-outer join defined as follows: Note that a diminisher is not a monoid.
The proof is given in Appendix A. For X = ⊗z, it implies that Y ⊗ Y = ⊗z.
Proof. Since ∆S ′ i ⊆ Si, then there must be Wi such that This theorem indicates that, to process the deletions ∆S ′ , we can simply use the same methods as for insertions, but using the diminishing function ⊗ for merging the state, instead of ⊗. That is, we can use the same functions h and a, but now the state is diminished as follows: For example, the join-groupby query used in Section 2 must now use the merging, X ⇓ (−)×(−) Y , equal to: select (θ , (sx−sy,cx−cy)) from (θ ,( sx ,cx)) in X , (θ ,( sy ,cy)) in Y union select x from x in X where π1(x) not in π1(Y )

IMPLEMENTATION
We have implemented our incremental processing framework using Apache MRQL [33] on top of Apache Spark Streaming [46]. The Spark streaming engine monitors the file directories used as stream sources in an MRQL query, and when a new file is inserted in one of these directories or the modification time of a file changes, it triggers the MRQL query processor to process the new files, based on the state derived from the previous step, and creates a new state.
We have introduced a new physical operator, called Incr, which is a stateful operator that updates a state. More specifically, every instance of this operation is annotated with a state number i and is associated with a state, statei, of type αi. The operation Incr(i, s0, F ), where F is a state transition function of type αi → αi and s0 is the initial state of type αi, has the following semantics: statei ← F (statei) return statei with statei = s0, initially. Note that the states are preserved across the Inc calls and are modified by these calls.
Recall that, in our framework, we break a query q into a homomorphism h and an answer function a, and we evaluate the query incrementally using: at every time interval, where, initially, state1 = ⊗z. This is implemented using the following physical plan: For an iteration repeat(λX. f (X, ∆S), n, g(S)), we use the approximate solution, described in Section 2: we split f into a homomorphism h, for some monoid ⊗, and a function a, and we derive a diffusion operator ⊗ : , ∆S))))

PERFORMANCE EVALUATION
The system described in this paper is available as part of the latest official MRQL release (MRQL 0.9.6). We have experimentally validated the effectiveness of our methods using four queries: groupBy, join-groupBy, k-means clustering, and PageRank. The platform used for our evaluations is a small cluster of 9 Linux servers, connected through a Gigabit Ethernet switch. Each server has 4 Xeon cores at 3.2GHz with 4GB memory. For our experiments, we used Hadoop 2.2.0 (Yarn) and Spark 1.6.0. The cluster frontend was used exclusively as a NameNode/ResourceManager, while the rest 8 compute nodes were used as DataNodes/NodeManagers. For our experiments, we used all the available 32 cores of the compute nodes for Spark tasks.
The data streams used by the first 3 queries (groupBy, join-groupBy, and k-means clustering) consist of a large set of initial data, which is used to initialize the state, followed by a sequence of 9 equal-size batches of data (the increments). The experiments were repeated for increments of size 10K, 20K, 30K, and 40K tuples, always starting with a fresh state (constructed from the initial data only). The performance results are shown in Figure 3. The x-axis represents the time points ∆t when we get new batches of data in the stream. At time 0∆t, we have the processing of the initial data and the construction of the initial state. Then, the 9 increments arrive at the time points 1∆t through 9∆t. The y-axis is the query execution time, and there are 4 plots, one for each increment size.   The join-groupBy and the k-means clustering queries are given in Section 2. The groupBy query is 'select (x,avg(y)) from (x,y) in S group by x'. The datasets used for both the groupBy and join-groupBy queries consist of pairs of random integers between 0 and 10000. The groupBy initial dataset has size 1M tuples, while the two join-groupBy inputs have size 100K tuples. The datasets used for the k-means query consist of random (X, Y ) points in 4 squares that have X in [2,4] or [6,8] and Y in [2,4] or [6,8]. Thus, the 4 centroids are expected to be (3,3), (3, 7), (7,3), and (7,7). This also means that the state contains 4 centroids only. The initial dataset for k-means contains 1M points and the k-means query uses 10 iteration steps. The k-means incremental program uses the approximate solution described in Section 2 based on approximate floating point equality.
From Figure 3, we can see that processing incremental batches of data can give an order of magnitude speed-up compared to processing all the data each time. Furthermore, the time to process each increment does not substantially increase through time, despite that the state grows with new data each time (in the case of the groupBy and join-groupBy queries). This happens because merging states is done with a partitioned join (implemented as a coGroup in Spark) so that the new state created by coGroup is already partitioned by the join key and is ready to be used for the next coGroup to handle the next increment. Consequently, only the results of processing the new data, which are typically smaller than the state, are shuffled across the worker nodes before coGroup. This makes the incremental processing time largely independent of the state size in most cases since data shuffling is the most prevalent factor in main-memory distributed processing systems.
We have also evaluated our system on a PageRank query on random graphs generated by the R-MAT algorithm [9] using the Kronecker graph generator parameters: a=0.30, b=0.25, c=0.25, and d=0.20. The PageRank query expressed in MRQL is a simple self-join on the graph, which is optimized into a single group-by operation. (The MRQL PageRank query is given in [18].) The PageRanks were calculated in 10 iteration steps. This time, the initial graph used in our evaluations had size 5K nodes with 50K edges and the four different increments had sizes 100 nodes with 100, 200, 300, and 400 edges, respectively. The performance results are shown in Figure 4. We believe that the reason that we did not get a flat line for our incremental PageRank implementation was due to the lack of sufficient memory tuning. In Spark, cached RDDs and D-Streams are not automatically garbage-collected; instead, Spark leaves it to  the programmer to cache the queued operations in memory if their results are used multiple times, and to dispose the cache later. This is reminiscent to legacy programming languages, such as C, that did not support GC. Removing all memory leaks is very crucial to continuous stream processing, and, if not done properly, it will eventually cause the system to run out of memory. We are planning to fine-tune our streaming engine to remove all these memory leaks.

CONCLUSION AND FUTURE WORK
We have presented a general framework for translating batch MRQL queries to incremental DSPE programs. In contrast to other systems, our methods are completely automated and are formally proven to be correct. In addition to incremental query processing on a DSPE platform, our framework can also be used on a batch distributed system to process data larger than the available distributed memory, by processing these data incrementally, in batches that can fit in memory. Although our methods are described using the unconventional MRQL algebra, instead of the nested relational algebra, we believe that many other similar query systems can use our framework by simply translating their algebraic operators to the MRQL operators, then, using our framework as is, and, finally, translating the resulting operations back to their own algebra.


Statistical Script Learning with Recurrent Neural Networks

Statistical Scripts are probabilistic models of sequences of events. For example, a script model might encode the information that the event “Smith met with the President” should strongly predict the event “Smith spoke to the President.” We present a number of results improving the state of the art of learning statistical scripts for inferring implicit events. First, we demonstrate that incorporating multiple arguments into events, yielding a more complex event representation than is used in previous work, helps to improve a co-occurrence-based script system’s predictive power. Second, we improve on these results with a Recurrent Neural Network script sequence model which uses a Long Short-Term Memory component. We evaluate in two ways: first, we evaluate systems’ ability to infer held-out events from documents (the “Narrative Cloze” evaluation); second, we evaluate novel event inferences by collecting human judgments. We propose a number of further extensions to this work. First, we propose a number of new probabilistic script models leveraging recent advances in Neural Network training. These include recurrent sequence models with different hidden unit structure and Convolutional Neural Network models. Second, we propose integrating more lexical and linguistic information into events. Third, we propose incorporating discourse relations between spans of text into event co-occurrence models, either as output by an off-the-shelf discourse parser or learned automatically. Finally, we propose investigating the interface between models of event co-occurrence and coreference resolution, in particular by integrating script information into general coreference systems.

Introduction
Natural language scripts are structured models of stereotypical sequences of events used for document understanding. For example, a script model may encode the information that from Smith landed in Beijing, one may presumably infer Smith flew in an airplane to Beijing, Smith got off the plane at the Beijing airport, etc. The world knowledge encoded in such event co-occurrence models is intuitively useful for a number of semantic tasks, including Question Answering, Coreference Resolution, Discourse Parsing, and Semantic Role Labeling.
Script learning and inference date back to AI research from the 1970s, in particular the seminal work of Schank and Abelson (1977). In this work, events are formalized as quite complex handencoded structures, and the structures encoding event co-occurrence are non-statistical and handcrafted based on appeals to the intuitions of the knowledge engineer. Mooney and DeJong (1985) give an early non-statistical method of automatically inducing models of co-occurring events from documents, but their methods are non-statistical.
There is a growing body of more recent work investigating methods of learning statistical models of event sequences from large corpora of raw text. These methods admit scaling models up to be much larger than hand-engineered ones, while being more robust to noise than automatically learned nonstatistical models. Chambers and Jurafsky (2008) describe a statistical co-occurrence model of (verb, dependency) pair events that is trained on a large corpus of documents and can be used to infer implicit events from text. A number of other systems following similar paradigm have also been proposed (Chambers and Jurafsky, 2009;Jans et al., 2012;Rudinger et al., 2015). These approaches achieve generalizability and computational tractability on large corpora, but do so at the expense of decreased representational complexity: in place of the rich event structures found in Schank and Abelson (1977), these systems model and infer structurally simpler events.
In this extended abstract, we will briefly summarize a number of statistical script-related systems we have described in previous publications (Pichotta and Mooney, 2016a;Pichotta and Mooney, 2016b), place them within the broader context of related research, and remark on future directions for research.

Methods and results
In Pichotta and Mooney (2016a), we present a system that uses Long Short-Term Memory (LSTM) Recurrent Neural Nets (RNNs) (Hochreiter and Schmidhuber, 1997) to model sequences of events. In this work, events are defined to be verbs with information about their syntactic arguments (either the noun identity of the head of an NP phrase relating to the verb, the entity identity according to a coreference resolution engine, or both). For example, the sentence Smith got off the plane at the Beijing air-port would be represented as (get off, smith, plane, (at, airport)). This event representation was investigated in Pichotta and Mooney (2014) in the context of count-based co-occurrence models. Balasubramanian et al. (2013), Modi and Titov (2014), and Granroth-Wilding and Clark (2016) describe systems for related tasks with similar event formulations.
In Pichotta and Mooney (2016a), we train an RNN sequence model by inputting one component of an event tuple at each timestep, representing sequences of events as sequences of event components. Standard methods for learning RNN sequence models are applied to learning statistical models of sequences of event components. To infer probable unobserved events from documents, we input observed document events in sequence, one event component per timestep, and then search over the components of a next event to be inferred using a beam search. That is, the structured prediction problem of event inference is reduced to searching over probable RNN output sequences. This is similar in spirit to a number of recent systems using RNN models for structured prediction (Vinyals et al., 2015;Luong et al., 2016;Dong and Lapata, 2016).
While the count-based event co-occurrence system we investigated in Pichotta and Mooney (2014) treats events as atomic-for example, the plane flew and the plane flew over land are unrelated events with completely independent statistics-this method decomposes events into components, and the two occurrences of the verb flew in the above examples have the same representation. Further, a lowdimensional embedding is learned for every event component, so flew and soared can get similar representations, allowing for generalization beyond the lexical level. Given the combinatorial number of event types, 1 decomposing structured events into components, rather than treating them as atomic, is crucial to scaling up the number of events a script system is capable of inferring. In fact, the system presented in Pichotta and Mooney (2014) does not use noun information about event arguments for this reason, instead using only coreference-based entity 1 With a vocabulary of V verb types, N noun types, P preposition types, and event tuples of arity k, there are about V P N k−1 event types. For V = N = 10000, P = 50, and k = 4, this is 5 × 10 17 . information.

System
Recall at 25 Human Unigram 0.101 -Bigram 0.124 2.21 LSTM 0.152 3.67 Table 1: Next event prediction results in Pichotta and Mooney (2016a). Partial credit is out of 1, and human evaluations are out of 5 (higher is better for both). More results can be found in the paper. Table 1 gives results comparing a naive baseline ("Unigram," which always deterministically guesses the most common events), a co-occurrence based baseline ("Bigram," similar to the system of Pichotta and Mooney (2014)) and the LSTM system. The metric "Recall at 25" holds an event out from a test document and judges a system by its recall of the gold-standard event in its list of top 25 inferences. The "Human" metric is average crowdsourced judgments of inferences on a scale from 0 to 5, with some post hoc quality-control filtering applied. The LSTM system outperforms the other systems. More results and details can be found in Pichotta and Mooney (2016a).
These results indicate that RNN sequence models can be fruitfully applied to the task of predicting held-out events from text, by modeling and inferring events comprising a subset of the document's syntactic dependency structure. This naturally raises the question of to what extent, within the current regime of event-inferring systems trained on documents, explicit syntactic dependencies are necessary as a mediating representation. In Pichotta and Mooney (2016b), we compare event RNN models, of the sort described above, with RNN models that operate at the raw text level. In particular, we investigate the performance of a text-level sentence encoder/decoder similar to the skip-thought system of Kiros et al. (2015) on the task. In this setup, during inference, instead of encoding events and decoding events, we encode raw text, decode raw text, and then parse inferred text to get its dependency structure. 2 This system does not obviously encode event co-occurrence structure in the way that the previous one does, but can still in principle infer implicit events from text, and does not require a parser (and can be therefore be used for low-resource languages).

System
Accuracy  More results can be found in the paper. Table 2 gives a subset of results from Pichotta and Mooney (2016b), comparing an event LSTM with a text LSTM. The "Copy/paste" baseline deterministically predicts a sentence as its own successor. The "Accuracy" metric measures what percentage of argmax inferences were equal to the gold-standard held-out event. The "BLEU" column gives BLEU scores (Papineni et al., 2002) for raw text inferred by systems (either directly, or via an intermediate text-generation step in the case of the Event LSTM output). The "1G P" column gives unigram precision against the gold standard, which is one of the components of BLEU. Figure 1, reproduced from Pichotta and Mooney (2016b), gives some example next-sentence predictions. Despite the fact that it is very difficult to predict the next sentence in natural text, the text-level encoder/decoder system is capable of learning learning some aspects of event cooccurrence structure in documents.
These results indicate that modeling text directly does not appear to appreciably harm the ability to infer held-out events, and greatly helps in inferring held-out text describing those events.

Related Work
There are a number of related lines of research investigating different approaches to statistically modeling event co-occurrence. There is, first of all, a body of work investigating systems which infer events from text (including the above work). Chambers and Jurafsky (2008) give a method of modeling and inferring simple (verb, dependency) pair-events. Jans et al. (2012) describe a model of the same sorts of events which gives superior performance on the task of held-out event prediction; Rudinger et al. (2015) follow this line of inquiry, concluding that the task of inferring held-out (verb, dependency) pairs from documents is best handled as a language modeling task.
Second, there is a body of work focusing on automatically inducing structured collections of events (Chambers, 2013;Cheung et al., 2013;Nguyen et al., 2015;Ferraro and Van Durme, 2016), typically motivated by Information Extraction tasks.
Third, there is a body of work investigating highprecision models of situations as they occur in the world (as opposed to how they are described in text) from smaller corpora of event sequences (Regneri et al., 2010;Li et al., 2012;Frermann et al., 2014;Orr et al., 2014).
Fourth, there is a recent body of work investigating the automatic induction of event structure in different modalities. Kim and Xing (2014) give a method of modeling sequences of images from ordered photo collections on the web, allowing them to perform, among other things, sequential image prediction. Huang et al. (2016) describe a new dataset of photos in temporal sequence scraped from web albums, along with crowdsourced story-like descriptions of the sequences (and methods for automatically generating the latter from the former). Bosselut et al. (2016) describe a system which learns a model of prototypical event co-occurrence from online photo albums with their natural language captions. Incorporating learned event co-occurrence structure from large-scale natural datasets of different modalities could be an exciting line of future research.
Finally, there are a number of alternative ways of evaluating learned script models that have been proposed. Motivated by the shortcomings of evaluation via held-out event inference,  recently introduced a corpus of crowdsourced short stories with plausible "impostor" endings alongside the real endings; script systems can be evaluated on this corpus by their ability to discriminate the real ending from the impostor one. This corpus is not large enough to train a script system, but can be used to evaluate a pretrained one. Hard coreference resolution problems (so-called "Winograd schema challenge" problems (Rahman and Ng, 2012)) provide another possible Input: As of October 1 , 2008 , OOV changed its company name to Panasonic Corporation. Gold: OOV products that were branded "National" in Japan are currently marketed under the "Panasonic" brand. Predicted: The company's name is now OOV .

Input:
White died two days after Curly Bill shot him.

Gold:
Before dying, White testified that he thought the pistol had accidentally discharged and that he did not believe that Curly Bill shot him on purpose. Predicted: He was buried at OOV Cemetery.

Input:
The foundation stone was laid in 1867.

Gold:
The members of the predominantly Irish working class parish managed to save £700 towards construction, a large sum at the time. Predicted: The OOV was founded in the early 20th century.

Input:
Soldiers arrive to tell him that OOV has been seen in camp and they call for his capture and death. Gold: OOV agrees . Predicted: OOV is killed by the OOV . alternative evaluation for script systems.

Future Work and Conclusion
The methods described above were motivated by the utility of event inferences based on world knowledge, but, in order to leverage large text corpora, actually model documents rather than scenarios in the world per se. That is, this work operates under the assumption that modeling event sequences in documents is a useful proxy for modeling event sequences in the world. As mentioned in Section 3, incorporating information from multiple modalities is one possible approach to bridging this gap. Incorporating learned script systems into other useful extrinsic evaluations, for example coreference resolution or question-answering, is another. For the task of inferring verbs and arguments explicitly present in documents, as presented above, we have described some evidence that, in the context of standard RNN training setups, modeling raw text yields fairly comparable performance to explicitly modeling syntactically mediated events. The extent to which this is true for other extrinsic tasks is an empirical issue that we are currently exploring. Further, the extent to which representations of more complex event properties (such as those hand-encoded in Schank and Abelson (1977)) can be learned automatically (or happen to be encoded in the learned embeddings and dynamics of neural script models) is an open question.