Current machine learning systems, such as TensorFlow and PyTorch, rely on high-performance linear algebra libraries for efficient tensor computations. Although they provide numerous fine-tuned array algorithms based on well-studied data placement and communication patterns, these libraries are hard to customize to capture irregular array programs and unconventional array storages. We present a framework for constructing distributed task workflows from ad-hoc tensor programs by partially evaluating these programs against the block coordinates of the tensors. In addition, we present a novel task scheduler based on pattern matching that assigns processes to tasks by recognizing certain patterns inside the task workflow. Although each such pattern applies to a small fixed number of tasks, when applied collectively, these patterns generate communication schemes that resemble optimal block-based algorithms, such as SUMMA. The tiling of the task workflow is based on pattern-matching and is done bottom-up, guided by cost.

Many vectorization languages and linear algebra libraries are often implemented as light-weight wrappers around high-performance array libraries, such as BLAS, and rely on the limited array storage structures and routines provided by these libraries, which are hard to extend and customize. In this paper, we describe a customizable framework for large-scale array programs in which arrays and array operations are abstract but their implementation is guided by user-defined storage mappings. We introduce a new storage structure for arrays, called a distributed tensor, which is a distributed collection of array blocks that may have any number of sparse and dense dimensions. In addition, we present rules for translating abstract array programs to high-performance distributed code that can run on Apache Spark. The performance of our system is on par with highly optimized linear algebra libraries, thus providing implementation independence and extensibility without sacrificing performance. Finally, we justify our claims by evaluating the performance of our system relative to Spark MLlib and TensorFlow.

Many graph algorithms can be expressed as repetitive computations that resemble matrix multiplication in which the addition and multiplication operations have been replaced with generalized operations that form an algebraic structure known as a semiring. Similar to matrix multiplication, these graph algorithms can be implemented in a distributed system using block arrays, which are distributed collections of non-overlapping dense arrays. In a distributed system, computations on block matrices are significantly faster than the same computations in the coordinate format in terms of computation and communication cost. Given the rise of Big Data and data-driven decision-making, our goal is to rewrite sequential programs on large graphs that are initially written to run on a single computer using loops and standard array operations to data-parallel programs on array blocks that can run on a distributed system. Our framework, called OSQLgen, automatically parallelizes graph programs with loops and arrays to distributed data-parallel programs. OSQLgen recognizes programs that are equivalent to a semiring structure and translates them to fast Spark SQL programs on block arrays that can run on a computer cluster in a distributed mode. We compare the performance of OSQLgen with GraphX, GraphFrames, and hand-written Spark SQL programs on coordinate and block arrays on various graph problems. On certain graph problems, OSQLgen is up to 36x faster than GraphX, 25x faster than GraphFrames, and up to 99x faster than hand-written Spark SQL programs on coordinate arrays, giving performance close to that of hand-written Spark SQL programs on block arrays.

Many programs written to analyze data are expressed in terms of array operations in an imperative programming language with loops. However, for data analysts who need to analyze vast volumes of data, large-scale data-intensive processing is becoming a necessity. Hence, they want to convert their programs, originally written to run on a single computer, to work on current Big Data systems, such as Map-Reduce and Spark, so that they can process larger amounts of data. We present a novel framework, called SQLgen, that automatically translates imperative programs with loops and array operations to distributed data-parallel programs. Unlike related work, SQL- gen translates these programs to SQL, which can be translated to more efficient code since it can be optimized using a relational database optimizer. SQLgen has been implemented on Spark SQL. We compare the performance of SQLgen with DIABLO, hand-written RDD-based, and Spark SQL programs on real- world problems. SQLgen is up to 78× faster than DIABLO and up to 25× faster than hand-written RDD-based programs, giving performance close to that of hand-written programs in Spark SQL.

Large volumes of data generated by scientific experiments and simulations come in the form of arrays, while programs that analyze these data are frequently expressed in terms of array operations in an imperative, loop-based language. But, as datasets grow larger, new frameworks in distributed Big Data analytics have become essential tools to large-scale scientific computing. Scientists, who are typically comfortable with numerical analysis tools but are not familiar with the intricacies of Big Data analytics, must now learn to convert their loop-based programs to distributed data-parallel programs. We present a novel framework for translating programs expressed as array-based loops to distributed data parallel programs that is more general and efficient than related work. We report on a prototype implementation on top of Spark and evaluate the performance of our system relative to hand-written programs.

This paper extends our previous work on deriving meaningful storm patterns from very large rainfall data. In an earlier work, we described MapReduce-based algorithms to identify three types of the storms: local, hourly and overall storms. In general, local storms have temporal characteristics of the storms at a particular site, hourly storms have spatial characteristics of the storms at a particular hour and overall storms have both spatial and temporal characteristics of the storm. We aim to find meaningful patterns and predict trajectories in the spatio-temporal data (i.e. overall storms which are sets of geographically overlapping, consecutive hourly storms). In this paper, we adapt K-Means clustering to find different types of hourly storms based on their shapes and sizes. Since the rainfall data are typically larger than the memory capacity of a single computer, we have implemented this clustering algorithm in Apache Spark, which is a distributed data processing framework, and have run our experiments on a computer cluster.

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, by retaining a minimal state during the query evaluation lifetime and by using a novel incremental evaluation technique, which, at each time interval, returns an accurate snapshot answer 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.

The MapReduce framework has become the de-facto framework for large-scale data analysis and data mining. One important area of data analysis is graph analysis. Many graphs of interest, such as the Web graph and Social Networks, are very large in size with millions of vertices and billions of edges. To cope with this vast amount of data, researchers have been using the MapReduce framework to analyse these graphs extensively. Unfortunately, most of these graph algorithms are iterative in nature, requiring repetitive MapReduce jobs. We introduce a new design pattern for a family of iterative graph algorithms for the MapReduce framework. Our method is to separate the immutable graph topology from the graph analysis results. Each MapReduce node participating in the graph analysis task reads the same graph partition at each iteration step, which is made local to the node, but it also reads all the current analysis results from the distributed file system (DFS). These results are correlated with the local graph partition using a merge-join and the new improved analysis results associated with only the nodes in the graph partition are generated and dumped to the DFS. Our algorithm requires one MapReduce job for pre-processing the graph and the repetition of one map-based MapReduce job for the actual analysis.

In our previous work, we described various aspects of our approach in converting big raw rainfall data into meaningful storm concepts. Three concepts were defined: local, hourly, and overall storms. The latter describes overall spatio-temporal characteristics of a storm as it progresses over time. We previously described MapReduce-based algorithms for local and hourly storm identification. Overall storms are the most complex to identify, and are at the core of the storm identification system. Multiple consecutive hourly storms that have spatial overlap are combined to create storm-centric characteristics of the whole storm, which could not be captured in most existing hydrology research. In this paper, we propose a MapReduce-based overall storm identification algorithm, which is based on iteration on MapReduce framework. This greatly improves performance when compared to the depth-first search (DFS) graph traveling approach as introduced in our previous work. In addition, additional essential storm characteristics of hourly and overall storms are introduced in this paper. Examples include storm center concepts for hourly storms and storm track and speed for overall storms.

We present an effective optimization framework for general SQL-like map-reduce queries, which is based on a novel query algebra and uses a small number of higher-order physical operators that are directly implementable on existing map-reduce systems, such as Hadoop. Although our framework is applicable to any SQL-like map-reduce query language, we focus on a powerful query language, called MRQL. Current map-reduce query languages, such as HiveQL and PigLatin, enable users to plug-in custom map-reduce scripts into queries for those jobs that cannot be declaratively coded in the query language, which may result to suboptimal, error-prone, and hard-to-maintain code. In contrast to these languages, MRQL is expressive enough to capture most of these computations in declarative form and at the same time is amenable to optimization. We describe an optimization framework that maps the algebraic forms derived from the MRQL queries to efficient workflows of map-reduce operations that consist of our physical plan operators. We also describe many algebraic optimizations, such as fusing cascading map-reduce jobs into one job and synthesizing a combine function from the reduce function of a map-reduce job. Finally, we report on a prototype system implementation and we show some performance results of evaluating MRQL queries on a small cluster of computers.

We address the problem of updating XML views over relational data by translating view updates expressed in the XQuery update facility to embedded SQL updates. Although our XML views may be defined using the full extent of the XQuery syntax, they can only connect relational tables through restricted one-to-many relationships that do not cause view side effects for a wide range of XQuery updates. Our approach is to use lineage tracing to propagate the necessary information about the origins of updatable data pieces through the query and the view code, to be used when these pieces are to be updated. Our system performs a compile-time analysis, based on polymorphic type inference and type usage, to detect the exclusive data sources, which are the table columns from the database that can be updated without causing side-effects to the view. The rest of the updates are associated with an update context in the form of a chain of tuples, which reflects the navigation path that was used to reach the update destination. At commit time, our system collectively considers all the compatible chains of all updates in the transaction and tries to relink them to new chains from the existing database whose tuples contain the updated data, so that the updates are reflected correctly without causing side effects to the other components of the view.

This paper introduces a framework for processing continuous, exact queries over continuous update XML streams. Instead of eagerly performing the updates on cached portions of the stream, we propagate the updates through the query evaluation pipeline, all the way to the result display, which prints the query answers. That way, the result display prints the query results continuously, replacing old results with new. The novelty of our approach is in the use of this processing framework to unblock operations and reduce buffering by letting the operations themselves embed new updates into the stream that retroactively perform the blocking parts of the operation. Based on this framework, we present novel methods for unblocking a number of important blocking/unbounded stream operations in XQuery using a small memory footprint, such as concatenation, general predicates, descendant-or-self and backward axes, and sorting.

We are presenting a framework for continuous querying of time-varying streamed XML data. A continuous stream in our framework consists of a finite XML document followed by a continuous stream of updates. The unit of update is an XML fragment, which can relate to other fragments through system-generated unique IDs. The reconstruction of temporal data from continuous updates at a current time is never materialized and historical queries operate directly on the fragmented streams. We are incorporating temporal constructs to XQuery with minimal changes to the existing language structure to support continuous querying of time-varying streams of XML data. Our extensions use time projections to capture time-sliding windows, version control for tuple-based windows, and coincidence queries to synchronize events between streams. These XQuery extensions are compiled away to standard XQuery code and the resulting queries operate continuously over the existing fragmented streams.

As most current query processing architectures are already pipelined, it seems logical to apply them to data streams. However, two classes of query operators are impractical for processing long or infinite data streams. Unbounded stateful operators maintain state with no upper bound in size and, so, run out of memory. Blocking operators read an entire input before emitting a single output and, so, might never produce a result. We believe that a priori knowledge of a data stream can permit the use of such operators in some cases. We discuss a kind of stream semantics called punctuated streams. Punctuations in a stream mark the end of substreams allowing us to view an infinite stream as a mixture of finite streams. We introduce three kinds of invariants to specify the proper behavior of operators in the presence of punctuation. Pass invariants define when results can be passed on. Keep invariants define what must be kept in local state to continue successful operation. Propagation invariants define when punctuation can be passed on. We report on our initial implementation and show a strategy for proving implementations of these invariants are faithful to their relational counterparts.

We are addressing the efficient processing of continuous XML streams, in which the server broadcasts XML data to multiple clients concurrently through a multicast data stream, while each client is fully responsible for processing the stream. In our framework, a server may disseminate XML fragments from multiple documents in the same stream, can repeat or replace fragments, and can introduce new fragments or delete invalid ones. A client uses a light-weight database based on our proposed XML algebra to cache stream data and to evaluate XML queries against these data. The synchronization between clients and servers is achieved through annotations and punctuations transmitted along with the data streams. We are presenting a framework for processing XML queries in XQuery form over continuous XML streams. Our framework is based on a novel XML algebra and a new algebraic optimization framework based on query decorrelation, which is essential for non-blocking stream processing.

The λ-DB project at the University of Texas at Arlington aims at developing frameworks and prototype systems that address the new query optimization challenges for object-oriented and object-relational databases, such as query nesting, multiple collection types, methods, and arbitrary nesting of collections. We have already developed a theoretical framework for query optimization based on an effective calculus, called the monoid comprehension calculus [4]. The system reported here is a fully operational ODMG 2.0 [2] OODB management system, based on this framework. Our system can handle most ODL declarations and can process most OQL query forms. λ-DB is not ODMG compliant. Instead it supports its own C++ binding that provides a seamless integration between OQL and C++ with low impedance mismatch. It allows C++ variables to be used in queries and results of queries to be passed back to C++ programs. Programs expressed in our C++ binding are compiled by a preprocessor that performs query optimization at compile time, rather than run-time, as it is proposed by ODMG. In addition to compiled queries, λ-DB provides an interpreter that evaluates ad-hoc OQL queries at run-time.

There is already a sizable body of proposals on OODB query optimization. One of the most challenging problems in this area is query unnesting, where the embedded query can take any form, including aggregation and universal quantification. Although there is already a number of proposed techniques for query unnesting, most of these techniques are applicable to only few cases. We believe that the lack of a general and simple solution to the query unnesting problem is due to the lack of a uniform algebra that treats all operations (including aggregation and quantification) in the same way.

We define a standard of effectiveness for a database calculus relative to a query language. Effectiveness judges suitability to serve as a processing framework for the query language, and comprises aspects of coverage, manipulability and efficient evaluation. We present the monoid calculus, and argue its effectiveness for object-oriented query languages, exemplified by OQL of ODMG-93. The monoid calculus readily captures such features as multiple collection types, aggregations, arbitrary composition of type constructors and nested query expressions. We also show how to extend the monoid calculus to deal with vectors and arrays in more expressive ways than current query languages do, and illustrate how it can handle identity and updates.