Nowadays, analytics are at the core of many emerging applications and can greatly benefit from the abundance of data and the progress in the processing capabilities of modern hardware. Still, new challenges arise with the extreme complexity of designing applications to exploit in full the potential of such oferings. In this work, we describe the main challenges in delivering advanced, cross-platform, streaming analytics at scale with the primary goal of interactivity. We present the architecture of a prototype system designed from scratch to tackle such challenges and we describe the internals of its components. Finally, we discuss key added-value synergies upon serving target analytics workflows in real-world applications and present a preliminary performance evaluation.
Interactive analytics at scale lie at the core of many diverse
applications. In life sciences, studying the efect of drug applications
on simulated tumors may produce data streams of 100Gb/min [
Relevant data not only originate from a variety of heterogeneous sources but also often arrive at multiple, geo-dispersed clusters. In life sciences, several eforts aim at federating analytics across Copyright © 2021 for the individual papers by the papers’ authors. Copyright © 2021 for the volume as a collection by its editors. This volume and its papers are published under the Creative Commons License Attribution 4.0 International (CC BY 4.0). Published in the Proceedings of the 2nd Workshop on Search, Exploration, and Analysis in Heterogeneous Datastores, co-located with VLDB 2021 (August 16-20, 2021, Copenhagen, Denmark) on CEUR-WS.org. data centers1. Similarly, discovering dependencies or correlations among economies or industries involves global analysis of stock streams originating from various local market data centers. In maritime data analysis, streams come from a number of ground- or satellite-based receivers before being ingested in proximate data centers. Additionally, Big Data technologies are significantly fragmented. Each of these scenarios, involves delivering analytics over networked clusters each hosting a variety of Big Data platforms.
Typically, such scenarios are designed as streaming analytics workflows. Our goal is to provide a solution for executing global, cross-platform, streaming analytics workflows over a network of computer clusters hosting diverse Big Data technologies, with high interactivity; i.e., continuously deriving the output of the worklfow in real-time and adapting to ad-hoc changes required due to observed results or system performance metrics, at runtime.
To this extent, streaming Big Data platforms such as Apache Spark, Flink, Kafka2 approach the problem by ensuring horizontal scalability, i.e, by scaling out (parallelizing) the computation to a number of Virtual Machines (VM) and providing APIs to program distributed Big Data processing pipelines. These platforms are not designed to support cross-platform execution scenarios. They also provide none or suboptimal support for online Machine Learning (ML) or Data Mining (DM) operators, since the major ML APIs they provide do not focus on streaming data. Frameworks such as Apache Beam or Apache SAMOA 3, allow for creating code that is executable in diferent, supported Big Data platforms. Nevertheless, this does not amend the lack of resource management in multicluster and cross-platform settings.
Related systems such as Rheem [
Our work aims at filling the gap in handling eficiently advanced,
cross-platform, interactive analytics at scale. We describe the
architecture of an open-source system, namely INFORE 4, designed from
scratch for this purpose. A recent INFORE demo [
real-world practitioners and applications, early in INFORE’s design, we identified three scalability requirements that should complement interactivity and cross-streaming platform execution features.
Enhanced Horizontal scalability. Parallelization is not adequate to handle extreme-scale scenarios. The number of available VMs in private clouds has a limit when it comes to extreme scale, while monetary charges arise for cloud providers.
Vertical scalability. Need to scale the computation with the number of processed streams. For instance, computing the correlation matrix of stock data streams results in a quadratic explosion in space and computational complexity, which is infeasible for thousands or millions of streams in our motivating scenarios.
Federated scalability. Need to scale the computation in settings
where data arrive at multiple, geo-dispersed sites. In such settings,
even when we do everything right within each cluster, the potential
for interactivity is network bound [
components for realizing the aforementioned key requirements.
Synopses Data Engine (SDE). There is a wide consensus in stream
processing [
The SDE builds and maintains such synopses across platforms
and clusters. It achieves enhanced horizontal scalability because
synopses shed the load assigned to each VM in a cluster and reduce
memory usage. For federated scalability, it reduces communication
by transmitting compact data summaries. For interactivity, we add
a unique SDE-as-a-Service (SDEaaS) design where the SDE is a
constantly running service (job, in one or more clusters) that can
accept any query, synopses creation on new streams or plugging
code for new synopses at runtime, with zero downtime for running
workflows. For vertical scalability, the SDEaaS design maintains
thousands of synopses for thousands of streams, in settings where
prior eforts [
Online Machine Learning and Data Mining (OMLDM). OMLDM
applies distributed, online ML and DM adopting a Parameter Server
(PS) paradigm [
Optimizer Component. The Optimizer is a prerequisite to efectively use the SDE and the OMLDM Components towards interactivity. The Optimizer picks (a) the networked cluster, (b) the Big Data platforms available in that cluster, and (c) the provisioned resources for each SDE, OMLDM (or other) operator of the global workflow. A good solution is determined taking into account performance criteria and cost models, related to interactivity, engaging throughput, (network and processing) latency, memory usage etc. These components cooperate as follows (more in Section 4).
Synopses-based Optimization. Given a workflow, the Optimizer produces alternative physical execution plans where it replaces exact (e.g., count, window, correlation) operators with equivalent, but approximate ones from the SDE. It then provides suggestions for equivalent, but approximate, workflows accompanied by the expected interactivity (execution speed up) vs. accuracy trade-of.
Boosting Vertical Scalability. The SDE Component boosts vertical
scalability of the OMLDM Component in ways that are not
possible otherwise. Indicatively, we use Discrete Fourier Transform
(DFT) and Locality Sensitive Hashing (LSH) bitmaps [
Pumped Parameter Server. With the help of the Optimizer, we
extend the OMLDM Component with an additional synchronization
policy, besides the synchronous and asynchronous
communication entailed by the classic PS design [
INFORE Architecture. We develop a prototype of the INFORE system incorporating operators from SDE, OMLDM and stream transformations provided by the streaming APIs of Big Data platforms such as Apache Flink or Spark. An overview of the INFORE architecture is illustrated in Figure 1(a). Currently, INFORE supports streaming analytics workflows over Apache Flink, Spark Structured Streaming and Kafka. Cross-(Big Data) platform and cluster communication is performed via JSON formatted Kafka messages. In a nutshell, INFORE works as follows:
(Step 1) The user designs in a Graphical Editor (left of Figure 1(a)) the desired workflow composed of stream transformation, OMLDM and SDE operators. These are logical operators composing the logical view of the workflow, independent of the underlying platforms. Logical WF
2
Suggested Manager Graphical Physical WF Editor
Stream OMLDM Transformations
Physical WF
Stats, Optimizer Logical WF
3 5 4 Monitor Submit
Computing Clusters Data Topic Request Topic
Map parse Map parse
CoFlatMap HashData Register Synopsis Register Request
FlatMap —Data Path —Requests Path
CoFlatMap
add estimate
Split splitter To geodispersed
Union Topic —Single-stream Synopsis Estimation
Union federator Union
Topic —Federated
Synopsis —Mergeable
Synopsis Estimation merge FlatMap (a) Overall INFORE Architecture. (b) SDEaaS Component (Condensed View).
Manager module
Statistics
Benchmarking Optimization Algorithms Cost Estimator
Training Topic Control Topic Prediction
Topic
Map parse
Map parse
FlatMap Learner
FlatMap Parameter
Server FlatMap Predictor
Feedback
Topic
Control Response Topic —Training
Path —Control
Path Prediction Output Topic —Prediction
Path (c) Optimizer Component (d) OMLDM Component (Condensed View). lab led tsaed lo o p G M U l edoM tree re
) iizannoo scyFanGM s t , llaboG raaPm rvSe rcySnh ,(scyn ltrsaaem c o l rrLaeen1 rrLaeen2 … rrLaeenn llsaendooM c o
L (e) Param. Server View. (Step 2) When the designed workflow is submitted, a JSON file is conveyed to a Manager module (middle of Figure 1(a)). Initially, the Manager sends the JSON to the Optimizer (right of Figure 1(a)).
(Step 3) The Optimizer determines the physical view of the worklfow, performing a mapping of the logical operators to physical ones to be executed on specific network clusters and Big Data platforms, so that various performance criteria are optimized.
(Step 4) The Optimizer returns a JSON with physical operators to the Manager which compiles . files deploying stream transformation, OMLDM and SDE operators (top of Figure 1(a)) to be submitted at chosen (cluster, Big Data platform) pairs.
(Step 5) The submitted jobs are monitored and performance statistics are collected.
This section elaborates on INFORE’s key components. For space considerations, we describe the concepts using Flink as an example implementation, but the design is generic to support operators in all streaming platforms mentioned above.
The architecture of the SDE Component is illustrated in Figure 1(b).
In streaming settings, multiple workflows may run continuously
providing output streams to application interfaces for timely
decision making. The SDEaaS design we adopt (see Section 2) can
serve multiple wofklows, running concurrently and continuously,
with approximate, commonly used operators (e.g., counts, sums,
frequency moments) reusing, instead of duplicating, streams and
synopses common to workflows. Moreover, for every new synopsis
maintenance request, SDEaaS reserves new execution tasks for the
operators in Figure 1(b), while without SDEaaS we would reserve
entire threads (task slots in terms of Flink). In that, SDEaaS manages
thousands of synopsis, on-demand, for thousands of streams; hence,
ensuring vertical scalability [
To briefly showcase the performance benefits of our approach,
Figure 3(a) shows the results of an experiment over a cluster with
40 task slots. (The hardware details are not important for this
experiment.) We compare SDEaaS to a non-SDEaaS approach, such
as Proteus [
The extensible SDE Library currently includes a variety of
synopses techniques for cardinality, frequency moment, correlation,
set membership or quantile estimation [
When a synopsis operator of the SDE is included in a workflow, every request arriving via the RequestTopic in Figure 1(b) for building or querying a running synopsis follows the red-colored path. The RegisterSynopsis, RegisterRequest FlatMaps initially produce and then look up the same keys pointing to workers that should maintain the synopsis, but for a diferent purpose each. RegisterSynopsis uses these keys to direct streaming tuples, which follow the blue-colored path, to assigned workers to update the synopsis. Register Request uses the same keys to direct queries to those workers and derive estimations at the estimation FlatMap by reading the status of the synopsis kept at the add FlatMap.
When a streaming update arrives at the Kafka DataTopic, the
FlatMap termed HashData looks up the keys of workers produced
by RegisterSynopsis and directs the update to the proper workers.
There, the add FlatMap includes the update in the maintained
synopsis. For instance, in case a FM sketch is maintained[
The rest of the operators on the right-hand side of Figure 1(b) are
used for appropriately controlling the output topics. The splitter
(split operator in Flink) directs the estimation to the OutputTopic
if a synopsis is defined on a single stream handled by one worker
(green path). If a synopses engages many streams (e.g. many stocks)
maintained at multiple workers (purple path) or multiple clusters
(yellow path – clusters communicate via the UnionTopic of Kafka)
merge FlatMap synthesizes the estimations (for instance, executing
a logical disjunction operation on various FM sketches [
The OMLDM component fills the gap of existing batch processingfocused APIs such as Spark’s MLlib or FlinkML. Its internal architecture is illustrated in Figure 1(d). Two types of pipelines are engaged in the OMLDM architecture: training and prediction pipelines.
Training pipelines. Training data streams follow the blue-colored path of Figure 1(d). They are ingested via the TrainingTopic. The rest of the training pipeline is implemented by a pair of Learner and ParameterServer FlatMaps. We follow a Parameter Server (PS) distributed model (Figure 1(e)), where an interactively defined number of identical learners distribute amongst themselves the computational load of training on an incoming data stream of training samples. In a nutshell, PS holds and updates the state of the learning process, including the current version of the trained model. Learners compute the updates to the model and coordinate via the PS. This is because the iterative nature of ML/DM computations requires communication between all parallel workers, and this is done by the PS in each pipeline. Training pipelines are data sinks, that is, they do not generate an output stream. However, they do have a Feedback Kafka topic to support iteration required by involved tasks. The Feedback topic is also read by the predictors of the red path, discussed below, to update their models upon a concept drift. For instance, should training on labeled streams of a classification task take place on par with prediction (tagging unlabeled streams), in case of a concept drift, the PS sends a new model to predictors via the FeedbackTopic.
For training pipelines, it is the pipeline’s responsibility to
coordinate the processing between the learners and the PS. As noted in
Section 2, the supported coordination policies include synchronous,
asynchronous and FGM, a novel and bandwidth-eficient
coordination policy that we introduced at [
Figure 1(d) shows an instance of the PS and learners running on a Flink cluster. The messaging interface between Learner and the PS is handled via Kafka, and hence it is generic and independent of the underlying Big Data platform.
Prediction pipelines. These are shown with the red-colored path of Figure 1(d). Prediction pipelines do not train the models they apply; these models are loaded at Predictor FlatMap via control messages. A model can be changed interactively by the PS at runtime, if there has been a concept drift, via the FeedbackTopic. Prediction pipelines employ some model for classification, interpolation or inference and transform input to output streams, which can be processed further by downstream operators of a global workflow. The control topics (green arrows and Kafka topics in Figure 1(d)) deliver control messages to and from an OMLDM job. The control API understands Create/Read/Update/Delete (CRUD) commands for each type of resource. For instance, using control messages the OMLDM component will respond with the current status of the PS and learners (e.g. how many learners are currently running) or the currently up-to-date model at the PS.
OMLDM currently supports (i) classification: Passive Aggressive Classifier, Online Support Vector Machines and Vertical Hoefding Trees, (ii) clustering/outlier detection: BIRCH, Online k-Means and StreamKM++ and (iii) regression: Passive Aggressive Regressor, Online Ridge and Polynomial Regression, Autoregressive Integrated Moving Average. Facilities for correlation matrix computation, standardization, polynomial feature extraction are also available. (a) Initial Workflow at the Graphical Editor. (b) Suggested Workflow via Synopses-based Optimization.
Given a workflow composed of various operators, we have to properly tune their execution prescribing both the Big Data Platform and the networked computer cluster that achieve good overall performance with respect to interactivity-related metrics are met.
The internal architecture of the INFORE Optimizer is shown in Figure 1(c). There are a number of submodules involved before providing logical to physical operator mappings to be deployed for execution. First, we use a statistics collector to derive performance measurements from each executed workflow. Statistics are collected via JMX or Slurm5 and are ingested in an ELK stack6 while monitoring jobs.
We have developed a Benchmarking submodule that automates the acquisition of performance metrics for every supported operator run on diferent Big Data platforms. The Benchmarking submodule circumvents steps 1-3 in Figure 1(a) and enables direct job submision to the underlying clusters engaging supported operators and stream sources. In that, the execution of various operators and (part of) workflows on diferent (Big Data platform, cluster) pairs is a priori benchmarked, statistics are collected and performance models are built to avoid a cold start.
Cost models are derived with a Bayesian Optimization approach
inspired by CherryPick [
Algorithmically, the Optimizer solves a multi-criteria optimization problem with objectives involving throughput, communication cost, processing and network latency, CPU/GPU and memory usage as well as accuracy for synopses-based optimization purposes. Due to conflicting objectives, we cannot optimize them all simultaneously. Therefore, we resort to Pareto optimal solutions obtained by maximizing a weighted combination of these objectives under clusters’ capacity constraints.
Notably, no prior efort [
This section describes how our prototype works in a real use case scenario from the financial domain and how advanced synergies are established among individual components. For exhibition purposes, we employ a single cluster equipped with Flink and Spark.
The logical workflow submitted to the Optimizer is illustrated in Figure 2(a). The workflow of Figure 2(a) utilizes Level 1, Level 2 stock data7 to discover highly positively/negatively correlated pairs of stocks. In Figure 2(a), data arrive at a Kafka source. The Split operator separates Level 1 from Level 2 data. It directs Level 2 data to the bottom branch of the workflow. There, the Level 2 bids are Filtered (i.e., for monitoring only a subset of stocks). The bids per stock are counted using a Count logical operator. When a trade for a stock is realized, a new Level 1 tuple is directed by Split to the upper part of the workflow. A Project operator keeps only the timestamp of the trade for each stock. The Join operator joins the stock trade, Level 1 tuple with the count of bids the stock received until the trade. The result is inserted in a time Window of recent such counts, forming a time series. Finally, the CorrelationMatrix of stocks’ time series is computed by that OMLDM operator.
ted, for every logical operator, there are two platform choices, Flink or Spark. The Optimizer holds a dictionary to determine the equivalence of logical operators to physical ones in either platform. In Synopses-based Optimization, the Optimizer attempts to boost interactivity by treating the SDE as an additional platform. That is, its dictionary also holds mappings about which exact operators could be replaced by approximate ones of the SDE. Therefore, the 7www.investopedia.com/terms/l/level1.asp,www.investopedia.com/terms/l/level2.asp Throughput versus Number of Synopses
Optimizer will finally suggest an execution plan as shown in Figure 2(b). In this plan the Count operator has been replaced by a SDE.CountMin sketch operator of the SDE and the Window operator has been replaced by SDE.DFT. All physical operators are prescribed to be executed over Flink, i.e., where the SDE service runs.
SDE, for every window it approximates, outputs a pair of values: a
bucketID and a vector of reduced dimensionality compared to the
original window. The bucketID is based on the first few coeficients
of the DFT [
Pumped Parameter Server. The stream entering the
Correlation Matrix operator of OMLDM is duplicated to the training
and prediction pipelines (Figure 1(d)). Learners (buckets in our
previous discussion) compute the pairwise correlations of the streams
they receive (correlation coeficient and the beta regression
parameter coincide for standardized vectors), while the PS holds the current
up-to-date correlation matrix which is essentially the global model
at the predictor and the PS sides. A concept drift for the global
model would occur if the Frobenius norm measuring the diference
between the matrix kept at PS and the true correlation matrix may
have exceeded a threshold. The Optimizer will decompose this
constraint to local ones installed at each learner. Mathematical details
are described in [
cluster with 4 Dell PowerEdge R320 Intel Xeon E5-2430 v2 2.50GHz machines with 32GB RAM and a cluster with 10 Dell PowerEdge R300 Quad Core Xeon X3323 2.5GHz machines with 8GB RAM each, for Flink and Spark Structured Streaming. We use real Level 1, Level 2 stock data (∼5000 stocks/∼10 TB) provided by a European ifnancial company. We set DFT to use 8 coeficients, CountMin ( = 0.002, = 0.01), we use 0.9 correlation threshold and a time window of 5 minutes, based on input from the data provider. Figure 3(b) shows the time it takes to the two optimization algorithms (Exhaustive and A*–alike) to produce a physical workflow from the time they receive the logical one, when 1 (Flink), 2 (Spark,Flink) or 3 (Spark, Flink, Synopses-based Optimization) are available. As the ifgure demonstrates for more than 1 platform, our novel A*–alike algorithm reduces the time to prescribe a physical workflow by 1 to 4 orders of magnitude. For 1 platform (still the Optimizer prescribes provisioned resources) the Exhaustive Search Algorithm is faster, since A*–alike requires more initialization time.
In Figure 3(c) we show the scalability of the INFORE approach versus a technique that executes the workflow of Figure 2(a) in Flink (denoted by "Parallelism"), without synopses-based optimization, boosted vertical scalability or the OMLDM Component. INFORE executes the workflow of Figure 2(b). We plot the throughput ratio of INFORE and "Parallelism" over a Naive approach that executes the same workflow with parallelism set to 1. The two horizontal axes in the figure show the number of processed stock streams (bottom) and the volume in terabytes (top) ingested using the maximum Kafka rate. Thus, the top axis accounts for horizontal scalability (volume, speed), while the bottom axis accounts for vertical scalability.
When we monitor 500 streams, i.e., 250K/2 cells in the correlation matrix, the fact that "Parallelism" lacks synopses-based optimization and vertical scalability makes its throughput ratio approaching 1, i.e. the Naive approach. INFORE’s performance remains steady when switching from 50 to 500 streams improving "Parallelism" by 3 times. The most important findings come upon switching to 5000 stocks (25M/2 cells in the matrix). Figure 3(c) denotes that because of the lack of enhanced horizontal and vertical scalability, "Parallelism" becomes equivalent to the Naive one. INFORE exhibits more than an order of magnitude improved throughput with the trade-of of highly correlated pairs being identified with 90% accuracy.
For federated scalability, we measure communicated bytes among the operators. Due to space constraints, we only report here that INFORE can reduce communication by more than an order of magnitude when 5000 stock streams are being monitored. 5
We have presented the design, internal structure, and integration aspects of key components of INFORE, a prototype designed for interactive analytics at scale. To our knowledge, INFORE is the first system for cross-platform, streaming analytics. We currently enrich the SDE and OMLDM libraries to strengthen the support for the three scalability types and enhance analytics capabilities.
This work has received funding from the EU Horizon 2020 research and innovation program INFORE under grant agreement No 825070.