Dataset Requirements.

To function on MRPack, dataset D should be formatted according to thebasic requirements of the algorithm. However, the type of data can be different; in unstructured text; the type can be a CSV, tab-separated, or space-separated. Similarly, for each algorithm, the dataset can be differentiated based on naming used for files in HDFS repositories. For example, when executing an InvertedIndex algorithm, its relevant dataset files in the repository can contain a descriptor to simplify the job of the preprocessor. In chained jobs, the output of one job is used as input of other jobs.

Data Preprocessor.

As discussed above, the dataset for MRPack is filtered being processed by the algorithms. We define a general preprocessor where the input data are preprocessed according to each algorithm requirements. A typical scenario of involving such data processing and filtering is described as below. In a given job J, each algorithm in A needs to process the subset of data from dataset D(D = d₁,d₂,d₃,…,d_n). Details of the algorithm are discussed in the next section. The preprocessed data are forwarded to the individual algorithm Map function where the Map task output is generated.

Intermediate Data Management.

Data communication plays a critical role in the performance of MapReduce. Data communication occurs in three stages i.e., data read and write to HDFS, from Map task to Reduce task (intermediate data), and between DataNodes (no-local data existence). These three data movements significantly affect the performance of the cluster. Here we describe our method for handling intermediate data to efficient management and processing.

Intermediate data generated by Map tasks belong to multiple algorithms in a job. These data are differentiable by the keys used by each algorithm and are specified in our hierarchical key scheme. In MRPack, custom comparator and partitioner functions are implemented to manage intermediate data. In generic MapReduce, these functions are optional; however, they are mandatory in our proposed scheme. The comparator function helps in algorithm specific data filtration and the partitioner function helps with skew mitigation. Based on the keys of each algorithm, we apply a skew mitigation strategy to avoid any delay in the job due to skewed data. We mitigate the data skew by partitioning intermediate data and forwarding it to specific Reducer. In MRPack, we use skew mitigation to generate separate output by each Reducer through comparator and partitioner algorithms. The details of comparator and partitioner algorithms are described in the algorithm description section.

Output Generation.

Efficient management of intermediate data simplifies output generation. To separate the output of each algorithm, we customize “data writer” and “MultipleOutputFiles” of Hadoop MapReduce, and write all algorithm’s data to a separate file. The separate outputs are necessary to differentiate between the results of each algorithm and reuse the data. These data can be easily reused in iterative algorithms where job chaining is required. The following section describes the algorithms.

Data Preprocessor Algorithm.

The preliminary part of MRPack algorithm is preprocessor. This part filters input data and differentiates between the subsets of data based on naming, data types, and structures. At the start of job J, this algorithm reads input from HDFS. This input data D is first parsed and processed in the preprocessor as shown in algorithm 1. This algorithm involves the data chunk from D which needs to be parsed against the specifications S of algorithms in set A(A = a₁,a₂,a₃,…,a_n). If d_i in D conforms to s_i in S; then, this data chunk is processed by that algorithm. Otherwise, it is skipped to parse the next data chunk. Each data chunk is processed by at least one algorithm and at most by all algorithms. The returned identified sub datasets are processed by individual Mappers of algorithms. This algorithm is invoked by main Mapper of job J.

Main Map Algorithm.

The Map function is the basic function of job in MRPack. In this function, the data preprocessor is first invoked to filter the data and return it to be processed by specific sub-Map as shown in algorithm 2.

This algorithm shows Mapper of the main job which consists of sub-mappers. A data chunk returned by the preprocessor is processed and key−value pairs are generated by one or many algorithms as shown by the functions Mapper_A lg₁,…,Mapper_A lg_n. These calling functions are individual Mappers of algorithms in MRPack. The sp in the algorithm indicates specific key configurations of each algorithm. For example, executing a machine learning algorithm such as k-nearest neighbor (KNN) requires a text dataset that consists of either comma or tab-separated fields. In this case, the preprocessor algorithm first identifies whether the data chunk is process-able and determines the format with defined algorithm specifications. These data are then processed by individual algorithm (e.g., kNN).

Algorithm 1: Preprocess Data (D = d₁,d₂,d₃,…,d_n)

 Data: D = d₁,d₂,d₃,…,d_n: Input datasets belonging to algorithms in A

   S = sp₁,sp₂,sp₃,…,sp_n: algorithmic specifications;

 Result: Algorithm specific labeled dataset

1 for ∀ d ∈ D do

2       /* check the data chunk */;

3  if D.d_i = = sp_i then

4   if Compliant(D.d_i) then

5        /* Check data compliance */;

6    D.d_i.sp = sp_i;

7      /* Filter the data for compliance */;

8   D.d_i.sp = Filter(D.d_i); D.d_i = D.d_i+1;

9  else

10     /* Set to null if the data is not for sp_i */;

11   D.d_i.sp = null; D.d_i = D.d_i+1;

12 return D.d_i   /* Return filtered and identified sub dataset */;

Algorithm 2: Map(D = d₁,d₂,d₃,…,d_n)

 Data: D = d₁,d₂,d₃,…,d_n: Input data chunks from HDFS

   S = sp₁,sp₂,sp₃,…,sp_n: algorithmic specifications;

 Result: Intermediate data: key-value pairs

1 for ∀ d ∈ D do

2     /* Preprocess the data before Mappers */;

3  D.d_i.sp = Preprocess(D.d_i);

4  if D.d_i.sp = = sp₁ then

5    /* Invoke individual algorithm Mappers from 1 to n */;

6   Mapper_Alg1(D.d_i);

7  if D.d_i.sp = = sp₂ then

8   Mapper_Alg2(D.d_i);

9  if D.d_i.sp = = sp₃ then

10   Mapper_Alg3(D.d_i);

11     /* continue till all algorithms are done */;

12  .

13  .

14  .

15  if D.d_i.sp = = sp_n then

16   Mapper_Alg_n(D.d_i);

17  D.d_i = D.d_i+1;

Main Reduce Algorithm.

The Reduce is the primary Reduce function of MapReduce. After pre-processing, Map output generation, and intermediate data shuffling and partitioning; it aggregates the data based on keys using specific Reduce function. Unlike traditional Reduce functions, this algorithm invokes a special function for a set of key-value pairs. The details are shown in algorithm 3.

Algorithm 3: Reduce(key,List[values])

 Data: Key−value pairs: Intermediate data key-value pairs

   K = k₁,k₂,k₃,…,k_n: Key formats of all algorithms;

 Result: Final aggregated result in Key-Value pairs

1 for ∀ value ∈ List do

2     /* Iterate over the incoming data */;

3  key = value.key; if key = = k₁ then

4    /* Invoke individual algorithm Reducer from 1 to n

5   Reducer_Alg1(value);

6  if key = = k₂ then

7   Reducer_Alg2(value);

8  if key = = k₃ then

9   Reducer_Alg3(value);

10     /* continue till all algorithms are done */;

11  .

12  .

13  .

14  if key = = k_n then

15   Reducer_Alg_n(value);

16  value = value.next();

In this algorithm, every time a Reducer receives a key−value list, it compares the key with metadata (algorithm specification) of algorithms and invokes a specific Reduce function. For example, the two algorithms in A, a₁ and a₂, have processed input data and generated Map output in the form of key−value pairs. The main Reduce algorithm receives and verifies the data with keys of all algorithms. If they belong to any algorithm, its Reduce function is invoked to collect the data and process it.

Sub−Map Algorithm.

The individual Map functions of each algorithm in MRPack function the same as those of generic MapReduce. However, unlike traditional MapReduce, sub-mappers function on part of the whole job (as job has multiple algorithms). The general sketch of a sub-Map is shown in the algorithm 4. Internal instruction set of each algorithm may vary; so we only shown the ConstructKey and GenerateValue as the computation functions.

Algorithm 4: sub-Map(d ∈ D)

 Data: d ∈ D data chunk: Data input preprocessed;

 Result: Emit data as (Key,Value) pairs

1    /* Construct key from dataset d ∈ D */;

2 key = Construct(d);

3 /* Process the data according to algorithm specific requirement */;

4 value = GenerateValue(d); /* emit the key and value as pairs */;

5 emit(key,value);

Sub−Reduce Algorithm.

Like sub-Map of MRPack, sub-Reduce also operates as sub-part of the main Reducer of an MRPack job and performs computation only on its own related data. Generally in a Reduce function, the intermediate data are aggregated and collected as shown in algorithm 5. However, in some special cases the data might require some computation such as kNN and K−Means requiring mean calculation and neighbors calculations in the Reduce function when implemented as part of this paper.

Algorithm 5: sub-Reduce(key,List < values >)

 Data: key−List < values > pairs: intermediate data;

 Result: Emit aggregated result as (Key,Value) pairs

1   /* Aggregate values based on same keys k ∈ K */;

2 for ∀ value ∈ List do

3     /* Iterate over the incoming data */;

4  if this.key = = key then

5   sum = sum + value;

6      /* Process data if needed */;

7  vals = ComputeVals(sum); /* Write the results */;

8  emit(key,vals);

Comparator and Partitioner Algorithm.

This part of the job plays an important role in many aspects such as intermediate data aggregation, sorting, partitioning, and data skew mitigation. The data are sorted by default by comparator based on keys. For partitioning and skew mitigation, we propose algorithm 6, which provides a brief sketch of partitioner. In this algorithm, the key is analyzed and then the value is forwarded to the appropriate Reducer. We set a threshold called λ to identify the skew mitigation. Suppose, we have three algorithms with keys as k₁,k₂, and k₃. If k₁ > 50%; then, the data of k₁ are diverted to another Reducer. Because the algorithm of k₁ is generating more intermediate data, which in turn can result in data skew and delay of the completion of the job. Selection of threshold value is based on experimentation. This value can be changed according to the cluster environment and specifications.

Key Structure and Hierarchy.

In MRPack, maintaining and managing different keys are difficult tasks. Data aggregation, partitioning, and sorting are all based on keys. To efficiently manage keys and overcome the challenge of skewed data, we design a hierarchical and composite key structure. In this scheme, the base class is a general abstract class, and for each algorithm, we extend the base class to use it for special case. We apply polymorphism and composition techniques to handle the keys. The general structure of the keys is shown in Fig 2. The key for each algorithm depends on its specification and requirements.

Download:

• PPT
  PowerPoint slide
• PNG
  larger image
• TIFF
  original image

Fig 2. Composite Key Structure: This structure shows keys modeling in MRPack where it is used to differentiate the algorithms.

https://doi.org/10.1371/journal.pone.0136259.g002

Algorithm 6: Partitioner(key,value)

 Data: key−List < values > pairs: intermediate data;

 Result: Partition data and Reducer allocation

1     /* Analyze keys and assign the reducer */;

2 keyFrequencyList = key; for i = 0ton do

3    /* Checks data skew if no skew, directly assign */;

4  if key = = k_i AND keyFrequencyList[key] < = λ then

5   return i%(numOfReducers);

6  if keyFrequencyList[key] > λ then

7  /* λ is a threshold set for maximum overloaded Reducer */;

8  for j = 0ton do

9     /* In case of data skew, check for least used Reducer */;

10   return leastIndexOf(keyFrequencyList[k_j]) % (numOfReducers);

Communication Cost.

Elapsed communication costs consist of data movement from Mappers (M) to Reducers (R) defined as in-memory data movement. Additional costs in a job are often incurred as result of failing, straggling tasks and data movement between DataNodes. The elapsed communication and additional costs for n and m number of M and R respectively is described as.

(1)

The total communication cost consists of elapsed, input-output, and additional cost. Therefore, (2) (3) where λ represents the rate of rescheduled straggling tasks in the cluster and σ represents the rate of rescheduled failing tasks with p and q representing their count respectively.

Computational Cost.

In an approximate assumption, an algorithm has complexity C executed over dataset D with a maximum number of Map tasks M_n and Reduce tasks R_m on a DataNode D_i (maximum number of tasks among all nodes). In MRPack, the job is first divided into Map and Reduce steps and then into individual algorithm Mappers and Reducers. Therefore, we estimate the computation cost as the maximum of the cost of Map i.e., max(Map) and Reduce i.e., max(Reduce).

(4)

For example, if max(Map) and max(Reduce) are O(n²) and O(nlogn) respectively, then (5) The total cost T_cost is finally given by the combination of communication and computation costs as: (6) (7)

This cost remains the same as that of a single MapReduce job, which demonstrates the significant improvement achieved through the use of MRPack. The complexity of MRPack for the set of algorithms is the same as that of MapReduce for a single algorithm.

Canadian government dataset.

The Canadian government dataset consists of data from the government’s different services including financial and national geographic information [27]. The data are released under the project name “Open Data”. We use the text and CSV format of the data from census of population category for our experimental purpose. This dataset is helpful in extracting word-based statistics even though it is not the focus of this paper.

Reuters-21578 and CV1.

This is the text dataset most widely used to measure the effectiveness of different systems and applications [28]. It includes datasets related to text categorization, corpus volume 1 (CV1) consisting of stories, and TREC-AP based on press articles. We use text data from [28] and do not perform any type of special information extraction techniques. However, to test the performance of MRPack, we process this dataset using the implemented algorithms and compare their performance to that of the generic MapReduce.

Machine learning dataset.

This is a publicly available dataset consisting of data from various fields including movies, songs, and companies [29]. This dataset is easily usable for machine learning experiments. We use this dataset mainly for processing by the machine learning algorithms such as kNN and KMeans.

Artificial dataset.

Volume is one of the foundations of the Big Data concept. Currently existing open datasets do not fulfill this requirement because they are very small. An artificial dataset is generated to fulfill the requirement of Big Data’s volume. Therefore, we generate both textual data and machine learning data based on the specification from the publicly available datasets. We perform experiments on varying datasets to accomplish scalability and performance analysis.

Cluster-size based analysis.

The results of the cluster size experiment are described in Fig 4. The total computation time for each size of cluster is calculated by the NameNode of the Hadoop cluster. The data size is maintained throughout the experiment, and nodes are added and removed after each test. The effect of cluster size is also measured by changing the data volume. With a constant data size at a certain threshold, the performance of the cluster improves by increasing the number of nodes as shown in Fig 4. In this case, when the data size is increased from 2GB to 8GB, the performance of the cluster improves with the addition of extra nodes. Similarly, the data size and cluster size are proportional to each other (i.e., in large datasets, addition of nodes decreases the total time).

Download:

• PPT
  PowerPoint slide
• PNG
  larger image
• TIFF
  original image

Fig 4. Cluster-based analysis of MRPack performance compared to that of generic MapReduce.

https://doi.org/10.1371/journal.pone.0136259.g004

Data-size analysis.

In this subsection, we evaluate the performance based on changing data size with constant cluster size. Both MRPack and MapReduce are executed on varying datasets and the same cluster consisting of eight nodes. In both cases, after certain threshold of datasets, the performance improved, as shown in Fig 5. However, the generic MapReduce executes a single algorithm in a single pass/job and MRPack executes multiple algorithms in a single pass/job. Explicit comparison between both with regard to data size show significant performance improvement for MRPack.

Download:

• PPT
  PowerPoint slide
• PNG
  larger image
• TIFF
  original image

Fig 5. Analysis based on changing data size w.r.t overall job execution time.

https://doi.org/10.1371/journal.pone.0136259.g005

Algorithm-based analysis.

We test the performance gain and loss when the number of algorithms in MRPack varies. In this case, we particularly monitor the I/O communication during job execution with HDFS. In MapReduce, each job requires to retrieve data from HDFS and write it back; hence, it results in long running jobs with heavy I/O operations as shown in Fig 6. However, in MRPack, the algorithms are executed as a single job and the I/O operations are performed only once. Hence, the performance is significantly affected and improved compared to that of MapReduce. In MapReduce, when we increase the number of algorithms, the number of jobs to be separately executed also increases. In MRPack, increasing the number of algorithms means changing the algorithms only in a single Job. By executing a single job, significant performance improvement is achieved, as shown in Fig 6. However, there are some memory-based limitations to this method, as discussed in the coming sections.

Download:

• PPT
  PowerPoint slide
• PNG
  larger image
• TIFF
  original image

Fig 6. Analysis based on number of algorithms in an MRPack job with respect to time execution time in terms of I/O and network communication.

https://doi.org/10.1371/journal.pone.0136259.g006

Memory stress analysis.

Here, we analyze the memory requirements and performance with respect to MRPack. Executing multiple algorithms in a single MapReduce job provides significant performance gain in I/O operations, data size, computation, and context switching between jobs. However, it also produces memory stress when the number of algorithms increases a certain limit or the memory of DataNodes is not sufficient. This problem can arise from frequently generated key-value pairs as intermediate data. This limitation can be solved by executing Map and Reduce tasks in parallel, where Reduce tasks process the output of completed Map tasks. The memory stress compared to that of other operations has a low impact on the overall job performance because of its in-memory data movement. I/O, computation, and context switching involve disk read and writes, thus providing a very efficient solution. Scalability of this approach depends on the memory architecture of the cluster. If machines in the cluster hold enough memory contain the intermediate data, then the scalability is the same as that of generic MapReduce. However, in other cases, the limitation of the number of algorithms in the jobs needs to be monitored.