A unified framework to improve the interoperability between HPC and Big Data languages and programming models

Future Generation Computer Systems 134 (2022) 123–139 Contents lists available at ScienceDirect Future Generation Computer Systems journal homepage: www.elsevier.com/locate/fgcs A unified framework to improve the interoperability between HPC and Big Data languages and programming models✩ César Piñeiro ∗,Juan C. Pichel Centro Singular de Investigación en Tecnoloxías Intelixentes (CiTIUS), Universidade de Santiago de Compostela, 15782 Santiago de Compostela, Spain article info Article history: Received 30 November 2021 Received in revised form 9 March 2022 Accepted 5 April 2022 Available online 11 April 2022 Keywords: Big Data HPC MPI Multi-language Programming models abstract One of the most important issues in the path to the convergence of HPC and Big Data is caused by the differences in their software stacks. Despite some research efforts, the interoperability between their programming models and languages is still limited. To deal with this problem we introduce a new computing framework called IgnisHPC, whose main objective is to unify the execution of Big Data and HPC workloads in the same framework. IgnisHPC has native support for multi-language applications using JVM and non-JVM-based languages. Since MPI was used as its backbone technology, IgnisHPC takes advantage of many communication models and network architectures. Moreover, MPI applications can be directly executed in an efficient way in the framework. The main consequence is that users could combine in the same multi-language code HPC tasks (using MPI) with Big Data tasks (using MapReduce operations). The experimental evaluation demonstrates the benefits of our proposal in terms of performance and productivity with respect to other frameworks. IgnisHPC is publicly available for the Big Data and HPC research community. ©2022 The Authors. Published by Elsevier B.V. This is an open access article under the CC BY license (http://creativecommons.org/licenses/by/4.0/). 1. Introduction The unification of High Performance Computing (HPC) and Big Data has received increasing attention in the last years. It is a common belief that exascale computing and Big Data are closely associated since HPC requires processing large-scale data from scientific instruments and simulations. But, at the same time, it was observed that tools and cultures of HPC and Big Data communities differ significantly [1]. One of the most important sources of divergence comes from the differences between their software ecosystems. In this way, HPC applications have traditionally been based on MPI (Message Passing Interface) to support inter-node parallel execution, and based on OpenMP or other alternatives to exploit intra-node parallelism. However, Big Data programming models are based on interfaces like Hadoop [2] or Spark [3]. In addition to different programming models, programming languages also differ between both communities: being Fortran and C/C++ the most common languages in HPC applications, and Java, Scala, or Python being the most common languages in Big Data applications. ✩This work has been supported by MICINN, Spain (RTI2018-093336-B-C21, PLEC2021-007662), Xunta de Galicia, Spain (ED431G/08, ED431G-2019/04 and ED431C-2018/19) and the European Regional Development Fund (ERDF). ∗Corresponding author. E-mail addresses: [email protected] (C. Piñeiro), [email protected] (J.C. Pichel). This divergence between programming models and languages sets out a convergence problem, not only related to the interoperability of the applications but also to the interoperability between data formats from different programming languages [4]. In this scenario, we need to consider how to build end-to-end workflows where, for example, simulations can be MPI applications written in Fortran or C/C++, and the analytics codes can be written in Java or Python (maybe parallelized by a Big Data framework). In this work we introduce a new computing framework called IgnisHPC1to deal with that issue. The main goal of IgnisHPC is to unify in the same framework the development, combination and execution of HPC and Big Data applications using different languages and programming models. With this objective in mind, we can summarize the main contributions of IgnisHPC as follows: – Unlike other frameworks such as Hadoop and Spark, IgnisHPC supports natively both JVM and non-JVM-based languages. Applications can be implemented using one or several programming languages following an API inspired by Spark’s one. –IgnisHPC uses MPI as backbone technology, which allows the framework to support many communication models and network architectures. In addition, MPI applications and libraries can be directly executed in an efficient way in IgnisHPC. In this way, most of the HPC scientific applications, 1It is publicly available at https://github.com/ignishpc. https://doi.org/10.1016/j.future.2022.04.002 0167-739X/©2022 The Authors. Published by Elsevier B.V. This is an open access article under the CC BY license (http://creativecommons.org/licenses/by/4.0/). C. Piñeiro and J.C. Pichel Future Generation Computer Systems 134 (2022) 123–139 which in many cases contain tens of thousands of lines of code, do not have to be ported to a new API or programming model. Nowadays, to the best of our knowledge, there is no other Big Data framework with that feature. – In IgnisHPC, MPI codes can be easily combined with typical MapReduce operations to create hybrid applications. Therefore, users can use the programming models that best fit their data-intensive and compute-intensive tasks in the same application. – A thorough experimental evaluation has been carried out to demonstrate the benefits of IgnisHPC in terms of performance and productivity. The study showed that IgnisHPC clearly outperforms Spark when considering different types of Big Data application patterns. Moreover, we proved that running MPI (and hybrid MPI+OpenMP) applications from IgnisHPC is easy and as efficient as executing them natively. – To avoid dependencies, IgnisHPC is fully containerized and supports some of the most well-known resource and scheduler managers. The remainder of this paper is organized as follows. Section 2 provides some context about Big Data and HPC technologies. Section 3explains in detail the architecture and modules of IgnisHPC. Section 4describes how to implement applications using the IgnisHPC API. Section 5focuses on the integration of MPI in IgnisHPC, and how MPI applications can be executed within the framework. The experimental evaluation is shown in Section 6. Section 7discusses the related work. Finally, the main conclusions derived from this work are explained. 2. Background 2.1. Big data frameworks MapReduce is a programming model introduced by Google for processing and generating large data sets on a huge number of computing nodes [5]. Apache Hadoop [2] was the first opensource implementation of the MapReduce programming model. It was widely adopted by both industry and academia, thanks to that simple yet powerful programming model that hides the complexity of parallel task execution and fault-tolerance from the users. However, most applications do not fit this model and require a more general data orchestration. Apache Spark [3] was designed to overcome some of the Hadoop limitations, especially when considering iterative jobs. Nowadays Spark is considered the de-facto standard for Big Data processing. Unlike Hadoop, Spark uses Resilient Distributed Datasets (RDDs) which implement in-memory data structures used to cache intermediate data across a set of nodes. Since RDDs can be kept in memory, algorithms can iterate over RDD data many times very efficiently. In addition, Spark provides many attractive features such as fault-tolerance, as RDDs can be regenerated through lineage when compute nodes are lost. Spark uses a thread-based worker model for executing the tasks. In this way, a Spark job is controlled by a driver program, which usually runs in a separate master node. On the other hand, the parallel regions in the driver program are shipped to the cluster to be executed. Spark is implemented in Scala and also has interfaces to execute Java, Python and R applications. Both Hadoop and Spark are capable of running codes written in other programming languages, but they suffer performance issues since they require sharing data outside the Java Virtual Machine (JVM) through system pipes [6]. Contrary to the common belief, this behavior also applies to Python in Spark because, while the Python driver code can be executed within the JVM thanks to Jython [7], executors are directly executed with the available Python interpreter. As a consequence, the performance of Python codes is affected like any non-JVM language such as C++ . In our previous work [8], we introduced Ignis, a first attempt to build an efficient and scalable multi-language Big Data framework. Despite being a prototype, Ignis has two significant contributions with respect to Spark. First, it allows to execute natively applications implemented using non-JVM languages such as Python and C/C++. Second, it supports multi-language applications, so different computing tasks could be implemented in the programming language that best suits them. However, there are also some important limitations as the following. For instance, Ignis is restricted to use TCP sockets for inter-node communication, so it is not possible to take advantage of typical HPC networks such as Infiniband, Aries [9], Slingshot [10], TofuD [11], etc. Another issue is that Ignis only supports one data partition per worker (executor). As a consequence, in order to work with big datasets, it is necessary to create new workers, which causes a degradation in the I/O performance. Ignis is completely containerized, but it uses a very limited ad-hoc solution for containers orchestration (named Ancoris). And finally, there was no submitter in Ignis, so jobs are manually launched using several configuration scripts. 2.2. MPI Message Passing Interface (MPI) is the most widely used and dominant programming model in HPC. In MPI, processes make explicit calls to library routines defined by the MPI standard to communicate data between two or more processes. These routines include both point-to-point (two party) and collective (many party) communication. Note that from MPI-3 new features were introduced to enable MPI processes within an SMP node to collectively allocate shared memory for direct load and store operations, which enables the shared-memory-processes to more efficiently share data. Among the different MPI implementations, the most successful ones are MPICH [12] and Open-MPI [13]. In particular, IgnisHPC uses MPICH as backbone technology to perform all the communications between processes, which allows the framework to support many communication models and network architectures (e.g. Infiniband, Slingshot, etc.). As we will explain later, one of the most important consequences of this design decision is that IgnisHPC can efficiently execute native MPI applications. Therefore, it is not necessary to port MPI applications using a different API. In this way, IgnisHPC is able to bring together the benefits of HPC and Big Data worlds into the same framework. We have considered MPICH instead of other MPI implementations because it is a mature project and provides several important features that are critical in a Big Data environment. For example, it is possible to join processes dynamically. It means that MPICH allows to connect independent instances of an MPI process at runtime, which is essential for increasing the number of processes when more parallelism is needed or for replacing lost processes when a computing node fails. 2.3. Resource managers and schedulers In a Big Data environment, it is necessary to manage and balance the cluster resources to allow multiple applications and frameworks to be efficiently executed together on the same system. That is the goal of resource managers such as Apache Mesos [14] and Nomad [15]. In particular, Apache Mesos groups all the physical resources of each node of the cluster and make them available for the applications as a single pool of resources. Among the features 124 C. Piñeiro and J.C. Pichel Future Generation Computer Systems 134 (2022) 123–139 Fig. 1. Scheme of the architecture of IgnisHPC. of Mesos we can highlight that it provides resource isolation thanks to its support to Docker containers [16]. So, it allows the execution of jobs in a custom independent environment both in terms of resources and installed software. Users cannot interact directly with Mesos because it is only a resource planner, so an orchestration framework is required to run and schedule tasks. We can find many orchestration frameworks depending on the type of tasks to be executed. In this work we have considered two of the most relevant, Apache Marathon [17] and Apache Singularity [18], which both support Docker containers orchestration. Finally, Nomad is a simple workload orchestrator created by HashiCorp. Its flexible and consolidated workflow provides users the functionality of both a resource and a scheduler manager, combined into a single system. Nomad is a modern alternative to Mesos and also supports containerized and non-containerized applications with different life cycle. Unlike old legacy platforms, it is prepared for running heterogeneous applications and using accelerators such as GPUs in a simple way. 3. IgnisHPC 3.1. Architecture of the framework We can divide the IgnisHPC architecture into four independent modules which run inside Docker containers: Submitter,Backend,Driver and Executor. A scheme is shown in Fig. 1. Modules were implemented in different languages, using Apache Thrift2 for the inter-module communications. In the figure, thin arrows represent those RPC communications, bold arrows data transfers, and numbers indicate the call order. The Submitter module is in charge of launching the IgnisHPC jobs (1) making a request to the resource manager (and scheduler), which is an external dependency that is responsible of the containers orchestration and the management of their resources. Afterwards, the driver container is started (2), where the Driver module is the container entrypoint. That module exposes all the available features of IgnisHPC through a user API created as interface to the Backend, where the API logic is implemented as a service (3). The Backend module is started inside the driver container after the driver code initializes the framework. Since Driver and Backend are in the same container, they share the same resources. The Backend is responsible of making the requests to the resource manager following the instructions specified in the driver code (4). As a consequence, the resource manager will create the executor containers (5). The 2https://thrift.apache.org Executor module contains the low-level implementation of a set of operations required by the Backend for each supported programming language. Note that IgnisHPC uses an SSH tunnel to connect driver and executor containers to encrypt and protect the communications. Finally, the Backend is connected to the executors in order to perform the low-level API operations (6). It is important to highlight that the Driver is also considered an executor by the Backend to handle data transfers. Note that only large data transfers are performed using MPI, otherwise RPC is used. There are important architectural differences between IgnisHPC and Ignis [8], our first prototype of a multi-language Big Data framework. Changes performed to IgnisHPC focused on removing some important limitations and performance issues shown by Ignis (see Section 2.1). Among them we can highlight the following: – One of the main goals of IgnisHPC is to unify the execution of Big Data and HPC workloads in the same framework. For this reason data transfers in IgnisHPC (bold arrows in Fig. 1) are performed using MPI. It has enormous advantages over the inter-node communications with TCP sockets used by Ignis. First, IgnisHPC supports many different communication models and network architectures (e.g. Infiniband, Slingshot, etc.). In this way, it covers the characteristics of the vast majority of Big Data and/or HPC clusters. Moreover, MPI applications and libraries can be directly executed in IgnisHPC. It means that HPC scientific applications, which in many cases contain tens of thousands of lines of code, do not have to be ported to the IgnisHPC API. And finally, it is possible to combine in the same multi-language code HPC tasks (using MPI) with Big Data tasks (using MapReduce operations). –IgnisHPC has a new Submitter module that handles jobs using an external resource manager. This module includes a submit script, similar to Spark’s spark-submit, that allows users to easily launch IgnisHPC jobs. On the contrary, jobs in Ignis are manually configured and launched using several scripts. There is no Submitter module in its architecture. – Ignis used a Manager module (one per executor container) that acted as middleman between the Backend module and the executors. To be more efficient, IgnisHPC removed that module and its functionalities were adopted by the Backend. –IgnisHPC supports some of the most well-known resource and scheduler managers. In addition, it was designed to easily add new managers. However, Ignis is bonded to work only with an ad-hoc manager, which has limited functionalities. 3.2. Jobs in the framework It is important to know the structure of an IgnisHPC job. It consists of a set of Docker containers distributed in multiple computing nodes grouped in Clusters (see Fig. 2). Workers are bonded to a single programming language and run inside a Cluster, so at least one Worker has to be created for each programming language in order to build multi-language applications. A Worker instantiates at least one process (executor) on each Docker container with the aim of executing its tasks in parallel, processing them as a pipeline. Clusters are independent, each one has its own assigned resources, so they can execute different tasks at the same time. Using multiple Clusters could be useful if stages or phases of the job have some compatibility issues. In this way, incompatible phases would be executed by differently configured Clusters. 125 C. Piñeiro and J.C. Pichel Future Generation Computer Systems 134 (2022) 123–139 Fig. 2. Job hierarchy in the IgnisHPC framework. However, Workers can be executed in shared mode (disabled by default). It means that executors of two or more Workers, which are located in the same container, would share the available resources. Normally users configure each Worker to use a part of the Cluster resources (e.g. cores, memory, etc.). 3.3. Resource manager Since IgnisHPC must be executed inside Docker containers, a resource and scheduler manager is required to handle the cluster resources and launch these containers. Note that any framework that meets these requirements could be used in IgnisHPC by implementing a basic interface. Currently IgnisHPC supports the following managers: –Docker: It is the easiest way to run IgnisHPC locally on a single machine. It uses the Docker client to directly launch containers. –Ancoris [8]: This is the only one supported by Ignis and it was designed as a simple and light resource manager. It executes itself inside Docker containers and is composed of two types of instances: master and slaves. The master manages the available resources in the cluster, and it is responsible of launching the containers. Slaves expose the resources of their host to the master when they are deployed. –Mesos+Marathon: Since Spark supports Mesos, using it as resource manager together with Marathon as container orchestrator allows users to execute in a single environment Spark and IgnisHPC jobs. In addition, since IgnisHPC is able to execute efficiently Big Data and MPI-based applications, we are merging both Big Data and HPC software ecosystems in just one execution environment. –Mesos+Singularity: The same benefits commented above apply to the combination of Mesos and Singularity. –Nomad: It combines in the same framework a resource and a scheduler manager. Due to its lack of dependencies, it is the best option to install in a cluster from scratch. Moreover, it has better support for devices like GPUs than Mesos, which allows to create heterogeneous execution environments. It is important to highlight that the cost of deploying Docker containers is very low. For instance, considering Nomad, it is possible to deploy thousands of Docker containers in just a few seconds.3 3https://www.hashicorp.com/c1m Fig. 3. Example of a task dependency graph. 3.4. Driver module The Driver module is a user API that allows access to all IgnisHPC functionalities. The driver program describes the highlevel control flow of the application, and it can be programmed in any of the supported languages (currently, Java, Python and C/C++). The Driver was designed as a Thrift RPC interface to the Backend so the logic has not to be re-implemented for every programming language. More details about the driver API and how to implement an application in IgnisHPC are provided in Section 4. 3.5. Backend module The Backend module contains the services that define the Driver’s logic. For instance, the reduceByKey function requires searching and grouping the keys. These operations are defined in the Backend, but they are implemented in the Executor module for a specific programming language. The Backend module is also responsible of sending requests to the resource manager in accordance with the instructions specified in the driver code. These instructions are lazily executed, so the Backend registers the function calls as a task dependency graph. When a task that represents an action in the driver code is created (e.g., a call to count), all the tasks in its dependency graph are executed. An example is shown in Fig. 3, where the Action Task depends on Task 3, and Task 3 depends on Task 1 and 2. Note that a task dependency is only computed if the task was never executed or if its result was not explicitly cached. The Executor and Container tasks are always executed as final dependencies. These tasks check that executors and containers are running and ready to be used. Finally, IgnisHPC is able to recover after a failure of a cluster node or some of the executors. Affected tasks are traced by the Backend in such a way that only their executors are reallocated and recomputed. If the affected tasks are cached, the recovery process will be faster since it is not necessary to recalculate their dependencies. Note that this process is automatic, but users can tune the recovery process using the persistence functions in the driver code (see Section 4). 3.6. Executor module The Executor module implements the operations defined by the Backend, where each supported programming language has its own implementation. In order to add support for a new language in IgnisHPC, a minimum implementation only requires programming the context class. The executor context allows the API functions to interact with the rest of the IgnisHPC system. In this way, among the functionalities of the context we find the exchange of user variables between driver and executors or providing the executors access to the MPI communicators. As we explained previously, IgnisHPC uses MPI (that is, MPI communicators) to perform all the communications related to the 126 C. Piñeiro and J.C. Pichel Future Generation Computer Systems 134 (2022) 123–139 Fig. 4. MPI communicators in IgnisHPC. Executor module. IgnisHPC constructs three types of communicators for data transfers (see Fig. 4): –Base communicator: for each worker there is a communicator that includes all its executors. This communicator always exists. If one executor is lost, the communicator is destroyed and a new communicator is created including a new executor. To that end, the capability of linking dynamically a single process to a communicator introduced in MPI-3 was of special importance. Without this feature all processes would have to be launched at the same time, and in case a process died, it could not be replaced causing the job to fail. –Driver communicator: it joins a base communicator to the driver process. It is created when the driver and a worker exchange data. –Inter-worker communicator: it is created joining the base communicators of two workers. It is used to send data from one worker to another and it will be destroyed as soon as one of the two workers stops its execution. This communicator is only created between workers that execute the operation ImportData. The base communicator for each worker is accessible to programmers by the executor context. It means that IgnisHPC functions can be implemented using that communicator and MPI primitives (e.g. gather, scatter, broadcast, reduce, etc.). As a result, IgnisHPC supports the execution of pure MPI applications with minimal modifications in the original code. A detailed explanation is provided in Section 5. Another benefit of using MPI for data transfers is the performance improvement of iterative applications. When using Big Data frameworks such as Ignis and Spark, an iterative application requires the driver to perform an evaluation task per iteration to obtain the final result. Each evaluation has three steps: stopping the executors, analysis of the partial results by the driver and restarting the executors. Note that starting and stopping the executors is very costly in terms of performance. IgnisHPC avoids the driver evaluations because executors share the partial results of each iteration using their MPI base communicator. Therefore, it is not necessary to stop them because they do not need to wait for the driver. This has even a bigger impact on performance for those applications with many short iterations. 3.7. Submitter module The Submitter is an IgnisHPC module consisting of a set of scripts and utilities for configuring and launching jobs. As we commented previously, Ignis had no module to launch tasks, and the Driver module was launched manually using Ancoris (see Section 3.3). The Submitter is a container, which can be accessed by SSH. There users can set up jobs in a similar environment where the IgnisHPC applications will be executed. The main utility of the Submitter module is the ignis-- submit script that, like spark-submit, allows users to launch Fig. 5. Job submission examples. IgnisHPC jobs in the cluster. The script only requires as mandatory arguments a Docker image and the driver program. There are also the following optional parameters: –name: a job name can be specified. –arguments: after the driver program name, all the parameters will be considered as driver arguments. –attach mode: by default, jobs are launched in unattached mode. That is, ignis-submit launches the job and exits. Attach mode allows users to control the job as if the driver runs locally, so output is printed in real time, and it is possible to manually kill the job. –properties: users can change the default properties before launching the Driver module. Executor properties can be redefined later but Driver properties are set only by ignis-submit. Fig. 5 shows two job submission examples. The first one is a Python basic submission with only its base image and the driver application. The second one deals with a C++ driver and optional parameters. In particular, --name sets the job name, --properties changes the driver default memory to 2 GB, and 0 -g 2 are considered arguments of mydriver. Note that ignishpc/cpp is the C++ base image. 3.8. Data storage IgnisHPC provides multiple options for data storage. Users can choose a type of storage according to their particular execution environment. Storage must be defined as a property before the worker creation. IgnisHPC supports the following storage options: –In-Memory: it is the best performer since all data is stored in memory. Memory consumption could be an issue, so it is not suitable for all kinds of jobs. –Raw memory: data is stored in a memory buffer using a serialized binary format. Extra memory consumption is minimal and the buffer is compressed by Zlib [19], which has nine compression levels. Level six is applied by default, but it can be changed when the worker properties are defined. –Disk: similar to raw memory but the buffer is stored as a POSIX file. Performance is much lower but it allows to work with large amounts of data that cannot be completely stored in memory. IgnisHPC behaves very similar to Spark in terms of data locality. Data is split into several partitions which are assigned to executors. Each partition is assigned to a single executor. In case another executor needs that partition, it will be sent using MPI. By default, IgnisHPC stores all partitions in memory to achieve the best possible performance. If there is not available memory, Docker sends some data to the container swap automatically. The user can modify this policy by changing the swap size or removing it. In case the data size exceeds the swap or the swap performance is insufficient, IgnisHPC allows users to set disk as primary storage, so partitions will always be stored on disk. In this case partitions will only be loaded into memory at processing time. 127 C. Piñeiro and J.C. Pichel Future Generation Computer Systems 134 (2022) 123–139 Table 1 Example of some IDataFrame functions supported by IgnisHPC. Type Functions Conversion map, filter, flatmap, keyBy, mapPartitions, keys, values, mapValues, etc. Group groupBy, groupByKey Sort sort, sortBy, sortByKey Reduce reduce, treeReduce, aggregate, treeAggregate, fold, reduceByKey, aggregateByKey, etc. I/O collect, top, take, saveAsObjectFile, saveAsTextFile, saveAsJsonFile, etc SQL union, join, distinct Math sample, sampleByKey, takeSample, count, max, min, countByKey, countByValue Balancing repartition, partitionBy Persistence persist, cache, unpersist, uncache On the other hand, there is an important difference in how memory is handled by Ignis and IgnisHPC. Since Ignis was just a prototype, for simplicity in the implementation, it assigns one data partition to each executor. In this way, if it is necessary to increase the partition size, a realloc operation is performed in such a way that the complete partition is copied to a different memory location. The consequence is a noticeable increment in the memory consumption. This restricts Ignis to work with smaller input datasets. IgnisHPC overcomes that limitation supporting several data partitions per executor. Note that an executor can spawn several threads to process the data partitions in parallel. 4. Programming applications for ignishpc IgnisHPC requires a minimal driver code that implements the application at high-level. To facilitate the adoption from the Big Data community, the IgnisHPC API is inspired by the Spark API in such a way that IgnisHPC codes are easily understandable by users who are familiar with Spark. In comparison to Ignis, we have extended the API to cover the most important primitives required by Big Data applications. For instance, IgnisHPC includes functions such as join and union for graph processing. The IgnisHPC driver API is composed by six main classes: –Ignis starts and stops the driver environment. –IProperties defines the execution environment properties. –ICluster represents a group of executors containers. It is possible, for example, to execute remote commands (execute,executeScript) and send files (sendFile, sendCompressedFile) to them. –IWorker represents a group of processes of the same programming language. This class includes functions to read files (textFile,partitionJsonFile, etc.), import data partitions from another worker (importData), send data from the driver (parallelize) and execute external codes (loadLibrary,call,voidCall). As we explain later, the former routines allow IgnisHPC to execute MPI applications within the framework. –IDataFrame contains all the functions of the MapReduce paradigm, similarly to Spark RDD. A function can be a transformation that generates another IDataFrame or an action that generates a final result (see Table 1). With respect to Ignis, besides the support for new API functions, IgnisHPC has increased the overall performance for some types of routines (e.g., Group and Sort) thanks to its complete redesign using MPI. Fig. 6. Transitive Closure driver code in Python. –ISource is an auxiliary class used by meta-functions such as map in the driver. This class acts as a wrapper for the input parameters. It is also used to store variables and send them to the executors. Those variables can be obtained by the executors using the context. Note that all the API operations that move data between executors perform an internal shuffling operation (e.g., parallelize,collect,partitionBy, etc.). 4.1. An example: Transitive closure With the goal of illustrating how applications are programmed in IgnisHPC,Fig. 6 shows an example of a driver implemented in Python for computing the Transitive Closure of a graph. This algorithm finds out if a vertex xis reachable from another vertex yfor all vertex pairs (x,y) in the graph. Note that an equivalent driver code could be implemented in any of the IgnisHPC supported languages (C/C++ and Java) using a similar syntax. First, the IgnisHPC framework is initialized (line 6). Properties are created to configure and build a cluster (lines 8 to 14). Note that the properties definition is optional, and IgnisHPC could read them from a default configuration file. Moreover, IgnisHPC introduce the possibility of overwrite the default values when a job is submitted like Spark using the new Submitter module. Therefore, the Docker image, the number of containers, the number of cores and the memory per container could be defined out of the driver code. Computing the Transitive Closure has two phases. To illustrate the multi-language support in IgnisHPC, each phase was implemented in a different programming language. The first one uses a Python executor and the second a C++ executor. The first stage consists of a map operation that takes as input a text 128 C. Piñeiro and J.C. Pichel Future Generation Computer Systems 134 (2022) 123–139 Fig. 7. Function in C++ used by a map operation for the Transitive Closure application. file and creates pair values that represent edges in the graph (line 20). As a consequence, it is necessary to previously create a Python worker in the cluster (line 16). It is important to note that creating the worker is mandatory and is not related to the driver programming language. On the other hand, if the worker and the driver code are in the same language, lambda functions can be used (line 20). The following phase of the algorithm is iterative: edges are joined into a path until there are no new paths. Since this phase is implemented in C++, a C++ worker should be created (line 22). Data is shared between workers using the importData function (line 24). The driver code ends printing the results. The framework must be stopped before the driver ends (line 41) to stop the backend. Unlike Ignis, IgnisHPC automatically detects when the driver process ends and stops it. However, this is not a good practice. In IgnisHPC a lazy evaluation is performed when a result is not required explicitly. In the example, the trigger that causes the tasks to be launched are the calls to the count function. This approach is also followed by Spark where RDDs are computed lazily the first time they are used in an action [20]. Most of the driver functions are meta-functions. That is, generic functions that require another one to perform an internal operation. This is the case of map in the example of Fig. 6 (lines 20 and 32–33). To implement those functions we should use the executor API provided by IgnisHPC. Basically, it defines a simple interface based on the number of required input parameters. Fig. 7 shows an example corresponding to the C++ function used by map in the driver code. Since map takes one parameter and also returns one parameter, IFunction is used. In case there are two input parameters (e.g., reduce), IFunction2 would be used, and so on. Note that if the function does not return any value (e.g. foreach), functions have the same name but with the Void prefix. 4.2. Text lambda functions As explained previously, lambda functions need that driver and executor codes were implemented in the same language because native code serialization is required. We refer to code serialization as the process by which a function or set of instructions are converted into bytes to be sent and executed in a different environment. Note that although Python and Java are able to serialize code both languages face compatibility problems. On top of that, C++ does not allow any type of code serialization. To overcome these limitations IgnisHPC implements its own multilanguage lambdas without source code serialization, named text lambdas. In this way, IgnisHPC allows to define lambdas as text, using the executor language syntax. The executor will transform the lambda text into source code to be used as a meta-function Fig. 8. Examples of text lambda for a Python and a C++ executor. parameter. It is important to highlight that the language of the driver code is indifferent. Fig. 8 shows an example of a text lambda that accumulates all elements (line 2) used by a reduce function. It uses Python syntax so must be evaluated by a Python executor. Another example is shown in line 7. It defines a text lambda that captures the value of a variable, which is read from the context. This lambda function will be compiled and loaded by a C ++ executor. Performance is not affected by using text lambda functions but it can add some overhead to the compilation process, especially for C++. In the same way, using the mechanism that allows to execute text lambda functions, IgnisHPC can send a complete job or application to the executors. This is possible thanks to loadLibrary, which can be used to execute a full source code as an IgnisHPC library. It has again a small impact on the compilation time. More details about loadLibrary are provided in Section 5.2 using MPI applications as use case. 5. MPI on IgnisHPC Our first prototype, Ignis, was limited to perform inter-process communications using only TCP sockets (different computing nodes) or shared memory (same node). However, IgnisHPC was completely redesigned to use MPI as backbone technology. As a consequence, all communications are internally implemented by MPI routines. As we explained previously in the paper, this change makes it possible for IgnisHPC to support more communication models and network architectures. In addition, an important advantage of our approach is that, once IgnisHPC has configured the MPI communications, users can combine in the same application pure MPI libraries using the IgnisHPC communicators together with standard Big Data functions (map,reduce, collect, etc.). 5.1. Integration of MPI into a big data environment MPI was not designed to run on Docker containers. As a consequence, there are several problems that should be addressed. First, by default and to preserve an isolation runtime environment, Docker creates a private virtual network between the host and the containers. If two containers are launched on the same host, we can execute MPI processes in the same way that they were two real nodes of a cluster. But if we launched those containers on different hosts, the communication is impossible since they belong to different networks. We can find in the literature several works that deal with this issue (see the Related Work section). For instance, some approaches opt for launching the container on the host network or creating a virtual network between the cluster nodes [21]. However, these configurations are difficult to handle and implement by resource managers in Big Data environments. Second, there are important differences in how ports are handled by a Big Data or an HPC environment. For instance, ports are considered a resource in a Big Data environment because there are services that require an exclusive port, which is not the case in 129 C. Piñeiro and J.C. Pichel Future Generation Computer Systems 134 (2022) 123–139 HPC. MPI needs ports to establish connections between processes but restricted to a range. However, ports provided by resource managers in a Big Data environment are usually random and not consecutive. Finally, IgnisHPC can internally spawn several threads per MPI process (executor) to increase the performance when processing and/or communicating data. All these threads use communicators to exchange data in parallel. Every time a communicator is created, MPI assigns a virtual interface to it. However, a virtual interface can only be used by one communicator at the same time, so parallel communications require the use of multiple virtual interfaces. In the most recent MPICH version, which is the MPI implementation used by IgnisHPC, virtual interfaces are assigned sequentially. Since IgnisHPC creates and destroys communicators dynamically, it is not possible to assure that threads can always exchange data in parallel using communicators with different assigned virtual interfaces. The consequence is a degradation in the performance. To overcome the above problems, IgnisHPC applies the following changes to MPICH: – Containers: MPICH has been designed to work on a local network. Docker containers can be joined to a network but only within the same node (internal network). Although resource managers can export a service from the internal network to the local network, this causes a problem when MPI is executed containerized. MPICH uses a service to store the network addresses of the launched MPI processes, but when using containers, each MPI process stores the values corresponding to the internal network which are not valid outside the node. For this reason, it is necessary to modify MPICH in order to store the correct network values that correspond to the local network. – Ports: now MPICH uses a list of ports provided by the resource manager instead of a range. – Multithreading: MPICH was modified to assure that all threads use a different virtual connection. In this way, communications between threads can always be performed in parallel. 5.2. Running MPI applications in IgnisHPC IgnisHPC can execute MPI applications implemented in any of the supported languages. We must highlight that, to the best of our knowledge, currently there does not exist another Big Data framework with this feature. Most of the MPI codes for HPC are implemented in C/C++, while applications in other languages such as Java and Python are a minority. So although IgnisHPC supports Python and Java, we will focus on C/C++ MPI applications. To explain how to execute an MPI application within IgnisHPC, we have considered LULESH [22] as guiding example. LULESH is a proxy HPC application for shock hydrodynamics with more than 5000 lines of C++ code. MPI applications, like other IgnisHPC codes, must be implemented using the executor API to be used from the driver (see Fig. 7). However, some minimal modifications should be previously applied to the original MPI codes: – MPI initialization: IgnisHPC controls the MPI environment, so MPI_Init and MPI_Finalize must be removed from the MPI application. –MPI_COMM_WORLD: MPI applications use a default communicator but IgnisHPC requires its own communicator. The simplest solution is to create a custom header to overwrite the default communicator. Fig. 9 shows an implementation of this functionality. Fig. 9. C header that replaces MPI_COMM_WORLD by IGNIS_COMM_WORLD (global variable). Fig. 10. Executor code for LULESH using C++. – I/O data: These modifications are optional. In some scenarios, it is interesting to allow IgnisHPC to handle the operations on input and output data of an MPI application. This is the case, for example, when the output file of the application will be afterwards processed by other IgnisHPC tasks. If IgnisHPC manages the output file, data is kept in memory. If not, the output file would be written to disk and read again to continue executing the following tasks, causing an important degradation in the performance. To do that, read and write functions related to input/output files will be removed from the MPI source code. As we explain next, they will be replaced by input and return parameters of the call function in the corresponding executor code. Fig. 10 shows the executor code for calling LULESH from the IgnisHPC driver. First, the global variable for the communicator of Fig. 9 is created (line 4) and initialized with the IgnisHPC MPI group (line 9). LULESH is a benchmark, so it does not receive any data from IgnisHPC. As a consequence, according to the executor API explained in Section 4, it is of type IVoidFunction0 (line 5). The only mandatory method to be implemented is call (line 130 C. Piñeiro and J.C. Pichel Future Generation Computer Systems 134 (2022) 123–139 Fig. 11. LULESH usage from a Python driver (top) and its equivalent C++ code (bottom). Fig. 12. Wordcount example as MPI hybrid application. 7). Inside that method, the application arguments are parsed from the IgnisHPC context. In our example, each argument is individually parsed to create a user friendly interface. However, the arguments could be parsed together as a list, reducing noticeably the necessary lines of code. The call method ends calling the LULESH main function (line 34). Afterwards, Lulesh class is exported (line 38). This operation is only necessary in C++. To finalize, an IgnisHPC library is created (line 40), which will be used to call LULESH from the driver. Finally, we show how to use an MPI application from the driver. The example of Fig. 11 focuses only in the necessary functions to execute LULESH using a Python and a C++ driver. Note that the MPI application should be previously compiled as a library (liblulesh.so). The example assumes that there is a C++ worker in the driver where two functions are executed: loadLibrary and voidCall. On the one hand, loadLibrary loads all the classes from the library declared in create_ignis_library. In this case, Lulesh is the only existent class. On the other hand, voidCall is an action that causes the execution of the library. If the library returns an output to IgnisHPC,voidCall should be replaced by call that would return an IDataFrame object. Library arguments in C++, which were parsed in the executor code, are added to the function using addParam from the ISource class. This syntax could be also used in Python. Nevertheless, keyword arguments in Python are a more elegant alternative (see line 2 in Fig. 11). 5.3. Hybrid applications In IgnisHPC an MPI code can be combined with typical MapReduce operations to create a hybrid application. In this way, the different computing tasks could be implemented in the programming model and language that best suits them. Fig. 12 shows a simple example of a Wordcount application where an MPI Python library is combined with IgnisHPC API functions. Input data is distributed and prepared by IgnisHPC (Tasks 1 and 2), so MPI is only responsible of the computeintensive part (Task 3). Observe that for using functions included in a Python library it is only necessary to load the library (line 5) and invoke the call routine with the name of the desired function (line 11). Finally, results are converted and written to disk in json format using the IgnisHPC API (Task 4). Fig. 13 shows another example. In this case, API functions are combined with Fig. 13. Sum of an array example as MPI hybrid application. explicit calls to MPI routines (line 9) with the aim of creating the hybrid application. 6. Experimental evaluation 6.1. Experimental setup The experiments shown in this section were carried out on a 12-node cluster, where each node consists of: – CPU: 2 ×Intel Xeon E5-2630v4 (2.2 GHz, 10 cores) – Memory: 384 GB of RAM – Storage: 8 ×4TB 7.2k SATA – Network: 2 ×10GbE It is a Linux cluster running CentOS 7 (kernel 3.10.0), Docker 20.10.2-ce and Spark 2.2.0 (with YARN [23] as cluster manager). Ignis and IgnisHPC run on an Ubuntu 20.04 image with MPICH 3.4.1. 6.2. Big data applications We have selected five workloads that represent different types of application patterns for which Spark is considered the best performing Big Data framework [24]: Minebench,TeraSort,K-Means, PageRank and Transitive Closure. Table 2 lists the use of the most important operators by each Big Data application, including basic core operators and specific ones implemented by the IgnisHPC framework. –Minebench (MB). This application4performs the calculation of SHA-256 hashes imitating the Proof-of-Work algorithm used in the Bitcoin protocol [25]. Do not confuse it with the data mining benchmark suite NU-MineBench. This algorithm has two phases which are implemented using two chained map operations. The first map is data-intensive, while the second is a compute-intensive task. In particular, in the first stage a set of Bitcoin transactions are grouped together forming a block proposal. A binary Merkle tree [26] is calculated for those transactions and its Merkle root hash is added to a block header. The second stage calculates the hash of the block header iteratively while the condition is not met. The strong scaling tests were obtained using an input file containing 300 K blocks (120MB), while the weak scaling experiments fixed the input data per core to 300 K blocks. 4Publicly available at: https://github.com/brunneis/minebench. 131 C. Piñeiro and J.C. Pichel Future Generation Computer Systems 134 (2022) 123–139 CRediT authorship contribution statement César Piñeiro: Software, Validation, Writing – original draft. Juan C. Pichel: Conceptualization, Methodology, Supervision, Writing – review & editing. Declaration of competing interest The authors declare that they have no known competing financial interests or personal relationships that could have appeared to influence the work reported in this paper. References [1] S. Heldens, et al., The landscape of exascale research: A data-driven literature analysis, ACM Comput. Surv. 53 (2) (2020). [2] T. White, Hadoop: The Definitive Guide, fourth ed., O’Reilly Media, Inc. 2015. [3] M. Zaharia, M. Chowdhury, M.J. Franklin, S. Shenker, I. Stoica, Spark: Cluster computing with working sets, in: Proc. of the 2nd USENIX Conf. on Hot Topics in Cloud Computing, HotCloud, 2010, p. 10. [4] M. Asch, et al., Big data and extreme-scale computing: Pathways to convergence-toward a shaping strategy for a future software and data ecosystem for scientific inquiry, IJHPCA 32 (4) (2018) 435–479. [5] J. Dean, S. Ghemawat, Mapreduce: Simplified data processing on large clusters, in: Symposium on Operating System Design and Implementation, 2004, p. 10. [6] M. Ding, et al., More convenient more overhead: The performance evaluation of hadoop streaming, in: Proc. of the ACM Symposium on Research in Applied Computation, 2011, pp. 307–313. [7] Jython, 2019, Online http://www.jython.org/. (Accessed April, 2019). [8] C. Piñeiro, R. Martínez-Castaño, J.C. Pichel, Ignis: An efficient and scalable multi-language big data framework, Future Gener. Comput. Syst. 105 (2020) 705–716. [9] B. Alverson, E. Froese, L. Kaplan, D. Roweth, Cray XC Series Network, White Paper WP-Aries01-1112, Cray Inc. 2012. [10] D. De Sensi, S. Di Girolamo, K.H. McMahon, D. Roweth, T. Hoefler, An in-depth analysis of the slingshot interconnect, in: Proceedings of the Int. Conf. for High Performance Computing, Networking, Storage and Analysis, SC, 2020. [11] Y. Ajima, et al., The tofu interconnect D, in: IEEE Int. Conference on Cluster Computing, CLUSTER, 2018, pp. 646–654. [12] MPICH, 2021, Online https://www.mpich.org. (Accessed October, 2021). [13] Open-MPI, 2021, Online https://www.open-mpi.org/. (Accessed October, 2021). [14] B. Hindman, et al., Mesos: A platform for fine-grained resource sharing in the data center, in: Proc. of the 8th USENIX Conf. on Networked Systems Design and Implementation, 2011, pp. 295–308. [15] HashiCorp, nomad: workload orchestration made easy, 2021, Online https: //www.nomadproject.io/. (Accessed October, 2021). [16] D. Merkel, Docker: lightweight linux containers for consistent development and deployment, Linux J. 2014 (239) (2014) 2. [17] Apache Marathon, https://mesosphere.github.io/marathon/. [18] Apache Singularity, https://getsingularity.com/. [19] J.T. Kukunas, V. Gopal, J. Guilford, S. Gulley, A. van de Ven, W. Feghali, High Performance ZLIB Compression on Intel Architecture Processors, Tech. rep. Intel, 2014. [20] M. Zaharia, et al., Resilient distributed datasets: A fault-tolerant abstraction for in-memory cluster computing, in: Proceedings of the 9th USENIX Conference on Networked Systems Design and Implementation, USENIX Association, 2012, p. 2. [21] M. de Bayser, R. Cerqueira, Integrating MPI with docker for HPC, in: IEEE Int. Conference on Cloud Engineering, IC2E, 2017, pp. 259–265. [22] I. Karlin, et al., Exploring traditional and emerging parallel programming models using a proxy application, in: 27th Int. Symposium on Parallel and Distributed Processing, 2013, pp. 919–932. [23] V.K. Vavilapalli, et al., Apache hadoop YARN: Yet another resource negotiator, in: Proc. of the 4th Annual Symposium on Cloud Computing, ACM, 2013, pp. 5:1–5:16. [24] J. Shi, et al., Clash of the titans: MapReduce vs. Spark for large scale data analytics, Proc. VLDB Endow. 8 (13) (2015) 2110–2121. [25] S. Nakamoto, Bitcoin: A peer-to-peer electronic cash system, 2008, http: //bitcoin.org/bitcoin.pdf. [26] R.C. Merkle, Protocols for public key cryptosystems, in: IEEE Symposium on Security and Privacy, 1980, p. 122. [27] X. Li, P. Lu, J. Schaeffer, J. Shillington, P.S. Wong, H. Shi, On the versatility of parallel sorting by regular sampling, Parallel Comput. 19 (1993) 1079–1103. [28] T.-S. Chua, J. Tang, R. Hong, H. Li, Z. Luo, Y. Zheng, NUS-WIDE: A real-world web image database from national university of Singapore, in: Proc. of the ACM CIVR, 2009, pp. 48:1–48:9. [29] J. Leskovec, A. Krevl, SNAP datasets: Stanford large network dataset collection, 2014, http://snap.stanford.edu/data. [30] X. Meng, et al., MLlib: Machine learning in apache spark, J. Mach. Learn. Res. 17 (1) (2016) 1235–1241. [31] R.S. Xin, J.E. Gonzalez, M.J. Franklin, I. Stoica, GraphX: A resilient distributed graph system on spark, in: 1st International Workshop on Graph Data Management Experiences and Systems, ACM, 2013. [32] I. Karlin, J. McGraw, J. Keasler, B. Still, Tuning the LULESH Mini-App for Current and Future Hardware, Tech. rep., 2013. [33] V.E. Henson, U.M. Yang, BoomerAMG: A parallel algebraic multigrid solver and preconditioner, Appl. Numer. Math. 41 (1) (2002) 155–177. [34] A. Sasidharan, M. Snir, MiniAMR - a Miniapp for Adaptive Mesh Refinement, Tech. rep., 2016. [35] S. Ghosh, M. Halappanavar, A. Tumeo, A. Kalyanaraman, A.H. Gebremedhin, Minivite: A graph analytics benchmarking tool for massively parallel systems, in: IEEE/ACM Performance Modeling, Benchmarking and Simulation of High Performance Computer Systems, PMBS, 2018, pp. 51–56. [36] J. Yang, J. Leskovec, Defining and evaluating network communities based on ground-truth, 2012, arXiv:1205.6233. [37] Y. Liu, B. Schmidt, D.L. Maskell, MSAProbs: multiple sequence alignment based on pair hidden Markov models and partition function posterior probabilities, Bioinformatics 26 (16) (2010) 1958–1964. [38] J. González-Domínguez, Y. Liu, J. Touriño, B. Schmidt, MSAProbs-MPI: parallel multiple sequence aligner for distributed-memory systems, Bioinformatics 32 (24) (2016) 3826–3828. [39] J. Mistry, et al., Pfam: The protein families database in 2021, Nucleic Acids Res. 49 (D1) (2020) D412–D419. [40] D. Wheeler, SLOCCount, 2021, Online http://www.dwheeler.com/sloccount. (Accessed November, 2021). [41] T. Adufu, J. Choi, Y. Kim, Is container-based technology a winner for high performance scientific applications? in: 17th Asia-Pacific Network Operations and Management Symp, APNOMS, 2015, pp. 507–510. [42] M.T. Chung, N. Quang-Hung, M.-T. Nguyen, N. Thoai, Using docker in high performance computing applications, in: IEEE 6th Int. Conference on Communications and Electronics, ICCE, 2016, pp. 52–57. [43] L. Benedicic, F.A. Cruz, A. Madonna, K. Mariotti, Portable, high-performance containers for HPC, 2017, arXiv:1704.03383. [44] A.J. Younge, K. Pedretti, R.E. Grant, R. Brightwell, A tale of two systems: Using containers to deploy HPC applications on supercomputers and clouds, in: IEEE Int. Conference on Cloud Computing Technology and Science, CloudCom, 2017, pp. 74–81. [45] P. Saha, A. Beltre, P. Uminski, M. Govindaraju, Evaluation of docker containers for scientific workloads in the cloud, in: Proc. of the Practice and Experience on Advanced Research Computing, 2018. [46] J. Higgins, V. Holmes, C. Venters, Orchestrating docker containers in the HPC environment, in: HPC. Lecture Notes in Computer Science, Vol. 9137, Springer Int. Publishing, 2015, pp. 506–513. [47] P. Saha, A. Beltre, M. Govindaraju, Scylla: a mesos framework for container based MPI jobs, 2019, CoRR abs/1905.08386 arXiv:1905.08386. [48] A. Gittens, et al., Matrix factorizations at scale: A comparison of scientific data analytics in spark and C+MPI using three case studies, in: IEEE Int. Conf. on Big Data, 2016, pp. 204–213. [49] J.M. Abuín, N. Lopes, L. Ferreira, T.F. Pena, B. Schmidt, Big data in metagenomics: Apache spark vs MPI, PLoS One 15 (10) (2020) 1–20. [50] M. Saxena, S. Jha, S. Khan, J. Rodgers, P. Lindner, E. Gabriel, Comparison of MPI and spark for data science applications, in: IEEE Int. Parallel and Distributed Processing Symposium Workshops, IPDPSW, 2020, pp. 682–690. [51] X. Lu, et al., High-performance design of hadoop RPC with RDMA over InfiniBand, in: 42nd Int. Conference on Parallel Processing, 2013, pp. 641–650. [52] M. Anderson, et al., Bridging the gap between HPC and big data frameworks, Proc. VLDB Endow. 10 (8) (2017) 901–912. [53] A. Gittens, et al., Accelerating large-scale data analysis by offloading to high-performance computing libraries using alchemist, in: Proc. of the 24th ACM SIGKDD Int. Conference on Knowledge Discovery & Data Mining, 2018, pp. 293–301. [54] N. Malitsky, Bringing the HPC reconstruction algorithms to big data platforms, in: NY Scientific Data Summit, NYSDS, 2016, pp. 1–8. [55] N. Malitsky, et al., Building near-real-time processing pipelines with the spark-MPI platform, in: NY Scientific Data Summit, NYSDS, 2017, pp. 1–8. [56] N. Malitsky, R. Castain, M. Cowan, Spark-MPI: Approaching the fifth paradigm of cognitive applications, 2018, arXiv:1806.01110. 138 C. Piñeiro and J.C. Pichel Future Generation Computer Systems 134 (2022) 123–139 César Piñeiro received his B.Sc. in Computer Science from the University of Santiago de Compostela (USC) and M.Sc. in Artificial Intelligence at Menéndez Pelayo International University (UIMP) organized in collaboration with the Spanish Association for Artificial Intelligence (AEPIA) in 2018. Currently, he is a Ph.D. candidate at CiTIUS. His main interests are focused on High Performance Computing (HPC) and Big Data. Juan C. Pichel received his B.Sc. and M.Sc. in Physics from University of Santiago de Compostela (Spain). In 2006 he received the Ph.D. in Computer Science from University of Santiago de Compostela. He was a visiting postdoctoral researcher at University Carlos III de Madrid (Spain) and University of Illinois at UrbanaChampaign (USA). He also worked as a researcher and project manager at Galicia Supercomputing Center (Spain). Currently he is an associate professor at University of Santiago de Compostela. His research interests include parallel and distributed computing, Big Data technologies, programming models and software optimization techniques for emerging architectures. 139