The increasing amount of data related to the execution of scienti c work ows has raised awareness of their shift towards parallel data-intensive problems. In this paper, we deliver our experience with combining the traditional highperformance computing and grid-based approaches for scienti c work ows, with Big Data analytics paradigms. Our goal was to assess and discuss the suitability of such dataintensive-oriented mechanisms for production-ready workows, especially in terms of scalability, focusing on a key element in the Big Data ecosystem: the data-centric programming model. Hence, we reproduced the functionality of a MPI-based iterative work ow from the hydrology domain, EnKF-HGS, using the Spark data analysis framework. We conducted experiments on a local cluster, and we relied on our results to discuss promising directions for further research.
Scienti c work ows are key tools in many research areas that rely on multiple, diverse, and distributed operations over various datasets, usually yielding signi cant computational complexity and data dependencies. Nowadays, the increasing amount of input, intermediate, and even output data related to the execution of work ows is shifting these originally computationally intensive systems towards parallel data-intensive problems. c 2017 Copyright held by the owner/author(s).
While current work ows rely on hundreds of gigabytes of
intermediate data [
In this context, scienti c work ows face new performance
and scalability challenges in terms of data management,
workload distribution, load balance, and scheduling, to name
a few. Given the data-intensive nature of these problems,
recent works have suggested the opportunity of combining the
traditional high-performance computing (HPC) and
gridbased approaches with Big Data (BD) analytics and
highthroughput (HTC) paradigms [
Given the data-intensive nature of these problems,
recent works have suggested the opportunity of combining the
traditional high-performance computing (HPC) and
gridbased approaches with Big Data (BD) analytics and
highthroughput (HTC) paradigms [
Our hypothesis is that BD techniques could be used to scale-up scienti c work ows, although the architectural differences between the analytics and scienti c worlds might require novel approaches to achieve satisfactory results. Therefore, in this work we aimed to assess the suitability of such data-intensive-oriented mechanisms for production-ready workows, especially in terms of scalability.
In previous works [
In this paper we address the suitability of such
dataintensive-oriented mechanisms for production-ready
workows. To study this, we reproduced the functionality of an
iterative work ow from the hydrology domain, EnKF-HGS
[
The goal of this paper is to present and discuss our experience with the application of those paradigms, platforms and infrastructures currently used in BD, to a scienti c workow. The rest of this paper contains: background on the BD techniques we relied on and their relation with scienti c work ows, and a high-level description of the work ow from the hydrology domain we considered as use case (Section 2); key architectural details of the proposed solution implemented in Spark (Section 3); the preliminary evaluations we conducted to study the behaviour of the resulting redesigned work ow (Section 4); and relevant highlights and directions for future work (Section 5).
Scienti c work ows are composed of heterogeneous and
coupled components that simulate di erent aspects of the
domain they model. These modules interact and exchange
signi cant volumes of data at runtime, hence making these
transfers e cient has a potential major impact in the overall
performance of the resulting application [
Data-centric analytics and BD tools like Hadoop and Spark
are being explored to provide straightforward data
distribution and caching mechanisms in pleasingly-parallel
dataintensive HPC applications. The inherently parallel nature
of these tools has resulted in positive experimental results
showing their suitability for massively parallel workloads like
MTC-like work ows [
However, this topic is still fairly new, and the
experience with applying these techniques is still limited. Previous
works have contributed with guidelines and methodological
approaches to make the design of scienti c work ows easier
and more e cient, with a user-centric and visual
perspective [
Major examples of these resource-intensive work ows are
multi-scale data analysis applications. The hydrology
domain is a representative example of the former. One of
1Apache Spark 1.6.0 documentation is available at
http://spark.apache.org/docs/1.6.0/
the state of the art simulators in this domain is the
EnKFHGS work ow [
As shown in Figure 1, HGS allows the numerical
simulation of all the relevant surface water and groundwater
processes in a pre-alpine type of valleys. Hence, each HGS
simulation in the ensemble of realizations represents a
longrunning compute-intensive process, which comprises the
sequential execution of two proprietary simulation kernels:
GROK and HydroGeoSphere (HGS) [
As described, the original work ow consisted of an MPI implementation of an EnKF, which relied on two legacy binaries to execute the simulation (GROK and HGS). EnKFHGS operates with a set of realizations, which constitute independent instantiations of the model, but with di erent parameters. They are simulated independently, and the output is gathered afterwards for further processing.
Since the work ow is iterative, we selected the popular data analysis tool Spark as representative of a BD programming model and execution engine. The most relevant design and implementation details of the nal implementation of the work ow in Spark are described in the following paragraphs. The procedures executed in the Spark driver process are identi ed using letters (from A to D), while numbers (1 to 6) refer to tasks that are computed distributively in the Spark executors.
The rst step is to load the necessary auxiliary les that every executor will need to properly run its data partition. This includes, for instance, the kernel binaries. Spark guarantees that these les will be available for the worker nodes in their current working directory.
Input data is read in the driver process in order to initialize the base model, composed of two main matrices, M1 and M2, in which each column c1;r and c2r corresponds to an instantiation, r, of the model. Additional data structures are created and initialised, and the parameters of the simulation are obtained.
Both matrices are distributed by columns in order to build the realization set, R. Each realization r is composed of the corresponding columns from both matrices, c1;r and c2;r, so that a data distribution process is needed to create the distributed dataset that will be transformed in the following stages and iterations. The rationale behind distributing the workload this way is that each realization can be simulated independently from the others, without any further communication. Additionally, we forced each partition to hold the data for a single realization in order to induce ne-grained parallelism. After realizations are distributed, the followoing steps are executed for each realization:
The GROK kernel writes the realization input data to a local le. HGS will read the realizations from this le in order to conduct the simulation of the model. With the input les from GROK, HGS simulates the model and writes its output for subsequent analysis. In steps 1 and 2 we must ensure that both binaries will be executed in the same node to exploit data locality. To achieve this, we run GROK and HGS in the same map function, which is an indivisible task in Spark. They thus act as an inner pipeline within the work ow.
The post-processing stage is partially distributed. First, the output from each HGS execution is read in each executor in order to create an updated realization set, R0. With this information we create a distributed matrix M10 , and conduct several distributed operations to avoid gathering the whole matrix in the driver.
Further operations with auxiliary matrices are executed in the driver in order to lter and randomize the input for the following iteration. The goal of this stage is to minimise the size of the dataset that needs to be collected in a the driver prior the model update. Note that to achieve this, signi cant data shu es must be executed.
Since not every stage of the analysis could be distributed, there is a step in which we aggregate a nal matrix that will be used to compute an update matrix. This matrix is distributed afterwards, so we can update the realizations without gathering the whole dataset in a single node.
The distributed update matrix is used to update every realization in parallel. The resulting realization set is persisted to the local storage of the nodes as a faulttolerance measure, and the following iteration starts.
After every iteration is executed, the output is stored to HDFS. This is executed in parallel, as every partition is stored independently. 4.
The goal of our evaluation is to assess the bene ts and drawbacks of the application of the techniques discussed in Sec. 2. We focused on absolute execution time and speedup to analyse the e ects of the memory and virtualization overheads of Spark. 4.1
In order to assess the performance and scalability of the application, we selected a local cluster as baseline. The speci cations and limitations of this testbed are described as follows.
This infrastructure comprised 11 slave nodes, with the speci cations shown in Tab. 1. Each slave node holds 8GB of RAM and two Intel Xeon E5405 @2.00GHz processors, with four cores each. In addition, an auxiliary node was necessary to host the driver process of the Spark implementation, which required 7GB in the largest experiment we conducted. This means that the container in charge of running the driver would require 7GB plus a 10% memory overhead (as con gured by default in the platform), 512MB extra memory for heap space, and other overhead sources like serialization bu ers. Since Spark adds signi cant memory overhead to drivers and executors, we had to add a larger node to bypass the memory constraints in the slave nodes. As a consequence, we added a node with an overall amount of 94GB of RAM and four Intel Xeon E7-4807 @1.87GHz processors, with six cores each.
Execution Models: Spark vs. MPI
The objective of these experiments is to detect the e ects that the execution models have on the performance of the work ow execution. We analysed the MPI-based implementation of EnKF-HGS and its data-centric version built in Spark, both running in the local cluster formerly described.
We allocated each Spark executor to one core in order to fairly compare scalability against single-core MPI processes. We run experiments increasing realization volumes and executors number. We measured the absolute execution time for a single execution (including the job launch time required by Spark) and computed the speed-up achieved. The results for those experiments are shown in Fig. 2, in which (a) and (c) correspond to MPI, and (b) and (d) correspond to Spark. Remarkably, Spark yields better execution times for every experiment, and its speed-up is better the larger is the experiment for a given number of workers. This might be a result of the redesign process. However, the speed-up in Spark for the largest experiment (i.e. 64 realizations on 64 executors) is lower than in the MPI case. The problem in this case is that the 64 executors cannot be scheduled at once due to their large memory requirements.
The main conclusion from this experiment is that, while the BD-inspired approach shows surprising performance results, the memory overhead of the execution framework hurts scalability, as less parallel executors can be allocated in the same infrastructure. As a result, the slimmer MPI processes seem more suitable for large scale execution of this work ow. Another interesting aspect is related to the post-processing stage of the work ow and its e ect on the overall execution time. At some point, with the growing number of the parallel executors, the post-processing computation becomes shorter in time than the data transferring time. Consequently, the post-processing stage starts to a ect the overall execution time more than with a fewer number of the parallel executors.
Scienti c work ows are becoming data-intensive, and their scale continues to grow with larger data volumes and problem complexities. There is a rising interest in exploiting the opportunity to leverage Big Data application models and infrastructures to increase work ow scalability. Nevertheless, the bene ts and drawbacks of these new tools have not been fully studied yet.
This work has explored the e ects these paradigms could have in current work ows. We aimed to detect their benets and drawbacks, and to extract from these knowledge a series of lessons and future research topics. We focused this work in two key elements to be compared: MPI and Apache Spark. We experimented with a speci c work ow from the hydrology domain, and analyzed its performance and scalability implemented on MPI and Spark and executed in a local cluster.
The main conclusion from the experiments made is that, while the BD-inspired approach shows surprising performance results, the memory overhead of the execution framework hurts scalability, as less parallel executors can be allocated in the same infrastructure. As a result, the slimmer MPI processes seem more suitable for large scale execution of this work ow.
While we considered this use case relevant for our objectives in terms of complexity and size, further experimentation with other work ows with di erent structures would be necessary to corroborate our results and conclusions. In addition, further analysis of the cost-performance trade-o of applying Big Data paradigms to work ows should be conducted, as well as a detailed analysis of the I/O-related overhead speci c to the work ows. 6.
This work has been partially funded under the Spanish Ministry of Economics and Competitiveness grant TIN201341350-P, the COST Action IC1305 \Network for Sustainable Ultrascale Computing Platforms" (NESUS), and the FPU Training Program for Academic and Teaching Sta FPU15/00422 by the Spanish Ministry of Education. We gratefully acknowledge Wolfgang Kurtz (IBG-3, Forschungszentrum JA~ ijlich GmbH) for providing the EnKF-HGS simulator source code. We also acknowledge Oliver Schilling (CHYN, University of Neucha^tel) for providing the hydrological model use-case.