Introduction

Scientists in many domains, such as bioinformatics, remote sensing, and physics, use Data Analysis Workflows (DAWs) to sequence activities involved in large-scale and complex scientific processes [1]; [2]. These DAWs are typically represented as a Directed Acyclic Graph (DAG), which consists of a set of tasks and some directed edges between the tasks; edges show data dependencies between tasks and the priority order of task execution.

Scientists often use heterogeneous cluster infrastructures to run their DAWs because of privacy and financial concerns. Heterogeneous clusters provide high-performance computing environments that enable efficient data analysis and the execution of large-scale DAWs in a reasonable amount of time [3]. DAWs are often executed on large amounts of data, resulting in long runtimes that can exceed days or weeks [4]; [5]; [6]. Thus, in such environments, the key objective is to schedule DAW tasks across computing resources in such a way that the total execution time, also known as the makespan, is minimized.

It is well known that a high degree of parallelization can be achieved in many DAWs by splitting the input data of individual tasks into chunks and processing them independently [7]. For example, in metagenomic DAWs, the size of a reference genome as frequent input data can vary from several KB to hundreds of GB, and the reference genome typically contains thousands of genome files. The reference genome can be divided into different bins of genome files and they are processed by several independent tasks in parallel. The main challenge here is how to partition the input data; what should be the appropriate size of each chunk of input data, and how should each task be assigned to a computing node so that the makespan is minimized. We call this problem task granularity. In heterogeneous environments, this challenge is aggravated because the computing power of the existing computing nodes is different, so choosing the input size of each task to be executed on each of these computing nodes is a very effective means in optimizing the makespan. Since each task of the DAW is equivalent to a job for the cluster when a workflow is submitted for execution, the terms task and job are used interchangeably in this study.

In this paper, we propose a novel approach to task granularity for metagenomic DAWs in cluster infrastructures with makespan minimization. We first formulate the problem as a mathematical model. We solve then the proposed model using the Genetic Algorithm method. Since the calculation of makespan requires a proper estimation of tasks runtime, we apply three different methods for this estimation and also compare their accuracy.

The paper is organized as follows: Section 2 presents a review of the related work, and Section 3 illustrates the problem statement. The proposed mathematical model is introduced in Section 4. Section 5 discusses solving the proposed model using genetic algorithm. Job runtime estimation is presented in Section 6, and experimental results are presented in Section 7. Finally, Section 8 provides concluding remarks and plans for further studies.

Related works

In this section, we first discuss data access patterns used in scientific workflows. We then cover the scheduling of scientific workflows on heterogeneous clusters. Finally, we focus on methods for predicting the runtime of tasks, as these estimates are often used as input for scheduling approaches.

Data access patterns used in scientific workflows

The data access patterns of workflow applications have been addressed by several studies [8]; [9]; [10]; [11]. Accordingly, the most commonly used patterns in scientific workflows are as follows (Fig. 1):

• Pipeline:This is the most basic and familiar pattern.

A set of computational tasks is chained in a sequence such that the output of a parent task is the input of its child task in the chain. Because of the line dependencies in a pipeline pattern, the execution of a task cannot begin until the execution of its parent has completed and it has received the data generated by its parent. • Scatter: An input data is divided into several chunks.

These chunks are distributed into multiple tasks (a bag of tasks), which can be executed simultaneously because there are no dependencies between them. • Gather: Multiple chunks of data are produced by multiple tasks. All of them are used as input data by a subsequent task. The later task may need to receive all the chunks and integrate them to start execution.

Pipeline Scatter Gather

Data Task Obviously, the scatter pattern can be so effective in reducing makespan in a distributed environment such as a cluster and cloud because it provides parallel execution of tasks on computing nodes. The implementation of scatter is an NPhard problem (See Section 3.1), so the user is not able to do it manually to achieve an acceptable makespan. In this study, we propose an approach to address this problem.

Scheduling of scientific Workflows on heterogeneous clusters

Generally, workflow scheduling on heterogeneous infrastructures can be done in two ways, statically or dynamically [12]; [2]. Static scheduling assigns tasks to compute resources in advance, assuming that accurate information about workflow and infrastructure resources is available. Dynamic scheduling doesn't require such assumptions. On the one hand, many heuristic and meta-heuristic approaches have been provided for this problem. HEFT [13] is considered to be the most famous of these. On the other hand, mathematical optimization approaches such as MILP [14] have proposed optimal solutions to this problem and have also analyzed the problem more extensively. However, the presented state-of-the-art scheduling approaches have in common that they already know the structure of the DAG and find an optimal or suboptimal assignment of tasks to the computing nodes. By addressing the scattering problem of a bag, our proposed approach not only provides a suitable structure for the bag and consequently for the DAG, but also finds the best scheduling of tasks to computing nodes with the objective of minimizing the makespan.

Task runtime prediction

Most of state-of-the-art techniques for workflow scheduling rely on accurate predictions of tasks runtime [15]. Therefore, the problem of predicting the runtime of scientific workflow tasks based on historical data has been studied extensively. In this research, the objective is to minimize the makespan. To compute the makespan value, an acceptable estimate of the runtime of jobs is required. Assuming that we have some historical data of execution traces of jobs on computing nodes, we use three different methods to predict the runtime of jobs (See Section 6).

Problem statement

This study addresses the problem of job/task granularity of scientific workflows in heterogeneous cluster environments with the aim of minimizing makespan. The case study is a metagenomic workflow where a reference genome containing a set of genome files is the main input data of the workflow. Building FM index (Full-text index in Minute space) over a reference genome (reference genome indexing) is a common and time-consuming task in metagenomic DAWs [21] and such a task is often used by the bioinformatics workflow community, so it is a good case study for optimizing job/task granularity.

In general, the system model includes the following steps: the first step is to collect a historical dataset of job execution traces on different computing nodes of the cluster. If this dataset is not already available, it can be collected by data sampling [3]. In the second step, a proper estimation of the job runtime and its memory consumption is performed using a prediction method. Then, by solving the proposed mathematical model, the optimal size of each chunk of input data for each job and also the assignment of jobs to computing nodes is obtained. Finally, the optimal job granularity and assignment is used to execute the DAW in the cluster.

A motivating example

Suppose there is a reference genome that contains five genome files of the following sizes: g 0 = 10, g 1 = 15, g 2 = 20, g 3 = 25, g 4 = 35. Moreover, assume that the cluster infrastructure has three computing nodes A, B and C. The runtime of a job with an input size of S on the computing nodes is calculated by the following functions: In Fig. 2, two different possible job building and assignments of the genome files is depicted. However, there are 3 5 = 243 different states as a solution among which the solution with the minimum makespan is the best.

• f A (S) = ln S 2 + 1 • f B (S) = ln S 2 + 2S • f C (S) = ln S 2 − 4S − 10

In a real example, Archaea1 has 488 genome files. Assuming that the number of cluster computing nodes is 10. Any approach based on complete enumeration and trial and error for assigning these genomes to computing nodes requires comparing a maximum number of 10 488 to different assignments. This approach is obviously impractical. It is noticeable that Archaea is a very small reference genome among the available reference genomes. Therefore, applying the above mentioned approach for solving the related assignment problem is not efficient or even applicable. The best approach to deal with this problem is to create a mathematical model for the problem and then apply available efficient algorithms for solving mathematical optimization models.

The proposed mathematical model

In a mathematical model, the objective function, decision variables and problem constraints are expressed in mathematical expressions. These models provide a deep insight into the structure of the problem [12]; [22]. They are therefore suitable not only for solving the problem using classical methods, but also for solving the problem using heuristic or meta-heuristic methods.

The input data for the mathematical model are described in Table 1. Also, this table explains the decision variables existing in the proposed model. In this section, the formulation of the problem constraints and the objective function of the proposed model are presented in detail.

Required memory for running jobs

This constraint states that the memory limitation of kth node must be met. This constraint must be done for all jobs and computing nodes.

mem k ≥ f mem ( n ∑ i=1 S i • y i j • x jk ) ∀ j ∈ {1, 2, ..., J}, ∀k ∈ {1, 2, ..., v}(1)

For each J j and CN k , ∑ n i=1 S i • y i j • x jk denotes the input size of J j on CN k and f mem estimates the memory required for J j .

Assigning jobs to cluster nodes

Constraints (2) and (3) imply that non-empty jobs must be assigned to exactly one node. Constrain (4) enforces that empty jobs cannot be assigned to any node.

1 n n ∑ i=1 y i j ≤ v ∑ k=1 x jk ∀ j ∈ {1, 2, ..., J}(2)v ∑ k=1 x jk ≤ 1 ∀ j ∈ {1, 2, ..., J}(3)v ∑ k=1 x jk ≤ n ∑ i=1 y i j ∀ j ∈ {1, 2, ..., J}(4)

Objective function

For constructing the objective function the following points should be highlighted:

• The objective function deals with the runtime of some bags of jobs on some computing nodes.

• If t k (Eq. ( 5))denotes for the above mentioned time for CN k then the total runtime (makespan) equals to max(t k ) 1 ≤ k ≤ v . The objective function aims to minimize this time (Eq. ( 6)).

• When more than one task is assigned to a node, the node wastes a certain amount of time between executing two jobs. ∑ J j=1 (x jk − 1) denotes that time.

• For each CN k , ∑ J j=1 f k (∑ n i=1 y i j • s i • x jk ) calculates the summation of the runtime of jobs assigned to CN k . ∑ n i=1 y i j • s i • x jk is the input size of J j , in which, f k () denotes an implicit function of that.

Runtime of jobs on each CN k is calculated by Eq.5.

t k = ( J ∑ j=1 x jk − 1) • st k + ( J ∑ j=1 f k ( n ∑ i=1 y i j • s i • x jk )) (5)

Therefore, the objective function of the model is as follows: min(max

(t k )) 1 ≤ k ≤ v (6)

This can be expressed as follows:

min α subject to:

α ≥ t k ∀k ∈ {1, 2, ..., v} α ≥ 0 Table 1

Models parameters/variables and their description.

Notation of parameters Description

CN = {node1, node2, ..., node v } Cluster node set v

The number of nodes of the cluster CN k kth node of the cluster mem k

Accessible memory size of CN k for running tasks st k Switch time between two jobs for CN k Gen = {g1, g2, ..., g n }

The set of genomes of the reference genome n

The number of genomes of the reference genome g i i th genome of the reference genome S i Size of g i Job = {J1, J2, ..., J J }

The set of jobs J j j th job J

The number of jobs

Notation of decision variables Description

x jk 1 iff J j is assigned to CN k , otherwise 0 y i j 1 iff g i is binned to J j , otherwise 0

Genetic optimization for solving the proposed model

Suppose there is a reference genome of a certain size with genome files g 1 , g 2 , . . . , g n . We want to group the genomes into a number of jobs and then assign the jobs to computing nodes CN 1 , . . . ,CN v of a cluster infrastructure.

It can be seen that the proposed model is a non-linear binary mathematical model. Due to the special structure of the constraints and the objective function of this model, linearizing it leads to a binary linear model with a significantly large number of constraints. Solving this model is very time consuming in terms of computation (it may even be impossible). On the other hand, in contrast to classic approaches, using genetic algorithms is a very powerful approach for treating discret models even if the model is nonlinear [23]. Based on this fact, using genetic approach can be an efficient approach for solving the proposed model without any linearization. In the following, we explain how to implement the presented model using a Genetic Algorithm (GA).

Chromosome structure

GAs mimic optimization during optimization by modelling genetic recombination and a fitness function. Hence, when using GA to solve a particular problem, the first concern to be addressed revolves around the determination of a suitable chromosome coding. Each chromosome represents the different parameters that characterize a solution to the problem.

[24]. In the solution, we consider a population of individuals, each individual being a potential solution to the problem described by the individual chromosome. The initial population is generated at random.

In this study, job granularity involves determining the assignment of genome files to jobs and the assignment of jobs to cluster nodes. Thus, a solution (chromosome) is a twodimensional array in which the indices indicate the genome file number. The elements of the first row contain the job numbers and the elements of the second row contain the computing node numbers. The representation of a chromosome in the GA implementation is shown in Fig. 3. As Fig. 3 shows g 1 and g 3 are assigned to J 1 while g 2 is assigned to J 4 . Moreover, J 1 and J 4 are scheduled to CN 3 and CN 2 , respectively.

GA operators

The crossover operator can help to inherit some chromosome fragments of excellent individuals to subsequent generations. In this study, the single-point crossover technique [23] is adapted during the performing of the crossover operator to produce new individuals. These new individuals are then assessed for their potential to contribute to the next generation of the population. An example is shown in Fig. 4. After crossover, the first new individual is not feasible to add to the next generation because J 5 has been assigned to two different computing nodes. The mutation operator is a technique that replaces some gene values with others to increase population diversity. We use swap mutation [25] to explore new regions of the solution space, where two positions on a chromosome are randomly selected and swapped. After mutation, the potential of new individuals to contribute to the next generation of the population is evaluated.

JOB RUNTIME ESTIMATION

In the GA algorithm, each individual should be assigned a value of the fitness score, which is shown in the objective function defined in Eq.5. Suppose the runtime of job j on node CN k is equal to t j . The challenge here is that there is no explicit function for calculating t j ; in other words, how long does it take to execute a job j of the size S j on a computing node CN k ? Therefore, we use the following different methods to predict the runtime of jobs on computing nodes, assuming that we have historical data from task execution traces:

• Linear Regression: Linear regression is a popular and simple machine learning algorithm that models the relationship between dependent and independent variables by analyzing and learning from current training results using a linear expression of independent variables [26]. • Polynomial approximation: According to mathematical theorems, any continuous function can be optimally approximated by a suitable polynomial [27]. Based on this theory, a polynomial of appropriate degree can be used to approximate the unknown function using the given historical data. In practice, lower degree polynomials are considered to avoid the socialisation of higher degree polynomials. Here we only consider degrees one, two and three. • Logarithmic of a polynomial approximation (logpol): As shown in [28], the curves of runtime according to data input size are logarithmic like and on the other hand, as mentioned above, the polynomials are a suitable method for estimation, so we use a combination of logarithmic and polynomial as a new approximation method.

These methods assume that there is a relationship between a task's input size and its runtime, using the input size as the independent variable. Thus, they can be used to predict task runtime for any task input sizes. Similar to related work, these methods use the size of the file on the hard disk as an input to their prediction models. In a heterogeneous cluster, a same task may have different run times on different computing nodes. Therefore, the methods create their prediction models for each computing node.

Experimental results

We developed a read-mapping workflow2 for metagenomic data in the popular workflow engine Snakemake [29]. The workflow was run on Allegro3 , a cluster infrastructure. We created a historical dataset from traces of the workflow execution on the cluster as a historical execution trace.

We selected and used three reference genomes with different sizes (small, medium and large) as input data of the workflow to perform the experiments. The specifications of the input data are described in Table 2. We also looked into different cluster sizes of 4, 8, 16, and 24 computing nodes in the experiments.

As previously stated, our approach consider the job granularity and scheduling problems simultaneously for a bag of task in a workflow while related works have addressed only the scheduling problem; Indeed, they have assumed that there are a number of jobs, each with a certain size, and they schedule these jobs to the cluster nodes so that the total runtime is minimized.

Most existing approaches use the file size on disk as the input for their predictions or approximate models. In [30] is a detailed discussion of why uncompressed input data size for compressed files should be used. Accordingly, we use the uncompressed file size to predict the runtime and the used memory of jobs.

Accuracy comparison of job runtime estimation methods

We compare the accuracy of the methods used to estimate job runtime (See Section 6). The linear regression was implemented using the sklearn.linear_model6 library. The polynomial and log-pol have been implemented using polyfit() from the Numpy7 library and Curve_fit() from the Scipy.optimize8 library, respectively. For comparison we use the Mean Absolute Percentage Error (MAPE). This metric is calculated using the Eq. 7 where A i is the actual value, F i is the predicted value and n is the number of fitted points.

MAPE = 1 n n ∑ i=1 | A i − F i A i |(7)

Figs. 5-7 show results for each input data separately with different cluster sizes. As shown in these figures, the polynomial method gives a more accurate estimate of the values of job run times than log-pol. Log-pol also outperforms linear regression.

Changing the number of jobs to improve makespan

One of the main challenges in this problem is that the number of jobs is not known in advance. The most obvious idea is to consider this number as equal to the number of genome files, i.e. n. This seems logical at first sight, since the possibility to consider empty jobs allows to potentially find any possible clustering for genome files in the form of jobs. However, from a practical point of view, this is inappropriate and impossible in most cases, because due to the large number of genome files, the number of decision variables and constraints increases significantly, making it impossible to solve

Conclusion and Future works

In this paper, an approach to the task/job granularity problem for metagenomic DAWs in cluster infrastructures with makespan minimization was proposed. The problem was first formulated as a mathematical model and then the proposed model was solved using the GA method. One of the main challenges in this problem is that the number of jobs is not known in advance. We overcame this challenge by adjusting the number of jobs as a factor of the number of computing nodes. For each increase in the number of jobs, the makespan is calculated. This procedure continues and evolves until the distance between two successive makespan values is negligible or insignificant from the decision maker's point of view.

Experimental results showed that a desirable makespan value can be obtained after a few steps of increasing the number of jobs. Furthermore, the calculation of makespan requires a proper estimation of the task runtime, so we applied three different methods for this estimation. Experimental results showed that the polynomial approximation outperforms. the future, we aim to generalize our proposed model so that it can be applied to other scientific domains. Since the proposed approach does not schedule the workflow, but optimizes a single step of the workflow, we intend to integrate it into a scheduling approach in the future work.