See discussions, stats, and author profiles for this publication at: https://www.researchgate.

net/publication/309055895

A Parallel Random Forest Algorithm for Big Data in a Spark Cloud Computing
Environment

Article in IEEE Transactions on Parallel and Distributed Systems · August 2016

DOI: 10.1109/TPDS.2016.2603511

CITATIONS READS
325 3,281

7 authors, including:

Jianguo Chen Kenli Li

Hunan University Hunan University, National Supercomputing Center in Changsha
174 PUBLICATIONS 4,694 CITATIONS 837 PUBLICATIONS 14,008 CITATIONS

SEE PROFILE SEE PROFILE

Zhuo Tang Kashif Bilal

Hunan University North Dakota State University
80 PUBLICATIONS 1,749 CITATIONS 95 PUBLICATIONS 3,332 CITATIONS

SEE PROFILE SEE PROFILE

Some of the authors of this publication are also working on these related projects:

Optimization View project

ML based caching in Edge Networks View project

All content following this page was uploaded by Jianguo Chen on 17 October 2018.

The user has requested enhancement of the downloaded file.

 1

A Parallel Random Forest Algorithm for Big Data

in a Spark Cloud Computing Environment
Jianguo Chen, Kenli Li, Senior Member, IEEE, Zhuo Tang, Member, IEEE,
Kashif Bilal, Shui Yu, Member, IEEE, Chuliang Weng, Member, IEEE, and Keqin Li, Fellow, IEEE

Abstract— With the emergence of the big data age, the issue of how to obtain valuable knowledge from a dataset efficiently and
accurately has attracted increasingly attention from both academia and industry. This paper presents a Parallel Random Forest (PRF)
algorithm for big data on the Apache Spark platform. The PRF algorithm is optimized based on a hybrid approach combining data-parallel
and task-parallel optimization. From the perspective of data-parallel optimization, a vertical data-partitioning method is performed to
reduce the data communication cost effectively, and a data-multiplexing method is performed is performed to allow the training dataset
to be reused and diminish the volume of data. From the perspective of task-parallel optimization, a dual parallel approach is carried out in
the training process of RF, and a task Directed Acyclic Graph (DAG) is created according to the parallel training process of PRF and the
dependence of the Resilient Distributed Datasets (RDD) objects. Then, different task schedulers are invoked for the tasks in the DAG.
Moreover, to improve the algorithm’s accuracy for large, high-dimensional, and noisy data, we perform a dimension-reduction approach
in the training process and a weighted voting approach in the prediction process prior to parallelization. Extensive experimental results
indicate the superiority and notable advantages of the PRF algorithm over the relevant algorithms implemented by Spark MLlib and
other studies in terms of the classification accuracy, performance, and scalability.

Index Terms—Apache Spark, Big Data, Cloud Computing, Data Parallel, Random Forest, Task Parallel.

1 I NTRODUCTION each one of them may contain a little information. The

1.1 Motivation second problem is to choose appropriate techniques that
may lead to good classification performance for a high-
W ITH the continuous emergence of a variety of new
information dissemination methods, and the rise of
cloud computing and Internet of Things (IoT) technologies,
dimensional dataset. Considering the aforementioned facts,
data mining and analysis for large-scale data have become
a hot topic in academia and industrial research.
data increase constantly with a high speed. The scale of
The speed of data mining and analysis for large-scale
global data continuously increases at a rate of 2 times every
data has also attracted much attention from both academia
two years [1]. The application value of data in every field
and industry. Studies on distributed and parallel data min-
is becoming more important than ever. There exists a large
ing based on cloud computing platforms have achieved
amount of worthwhile information in available data.
abundant favorable achievements [3, 4]. Hadoop [5] is a
The emergence of the big data age also poses serious
famous cloud platform widely used in data mining. In [6, 7],
problems and challenges besides the obvious benefits. Be-
some machine learning algorithms were proposed based on
cause of business demands and competitive pressure, al-
the MapReduce model. However, when these algorithms
most every business has a high demand for data processing
are implemented based on MapReduce, the intermediate
in real-time and validity [2]. As a result, the first problem
results gained in each iteration are written to the Hadoop
is how to mine valuable information from massive data
Distributed File System (HDFS) and loaded from it. This
efficiently and accurately. At the same time, big data hold
costs much time for disk I/O operations and also massive
characteristics such as high dimensionality, complexity, and
resources for communication and storage. Apache Spark
noise. Enormous data often hold properties found in various
[8] is another good cloud platform that is suitable for data
input variables in hundreds or thousands of levels, while
mining. In comparison with Hadoop, a Resilient Distribut-
ed Datasets (RDD) model and a Directed Acyclic Graph
• Jianguo Chen, Kenli Li, Zhuo Tang, and Keqin Li are with the College of
Computer Science and Electronic Engineering, Hunan University, and (DAG) model built on a memory computing framework is
the National Supercomputing Center in Changsha, Hunan, Changsha supported for Spark. It allows us to store a data cache in
410082, China. memory and to perform computation and iteration for the
Corresponding author: Kenli Li, Email: [email protected].
• Kashif Bilal is with the Qatar University,Doha 2713, Qatar, and the
same data directly from memory. The Spark platform saves
Comsats Institute of Information Technology, Islamabad 45550, Pakistan. huge amounts of disk I/O operation time. Therefore, it is
• Shui Yu is with the School of Information Technology, Deakin University, more suitable for data mining with iterative computation.
Melbourne, Vic.3216, Australia. The Random Forest (RF) algorithm [9] is a suitable data
• Chuliang Weng is with the School of Computer Science and Software
Engineering, Institute for Data Science and Engineering, East China mining algorithm for big data. It is an ensemble learning
Normal University, Shanghai 200241, China. algorithm using feature sub-space to construct the model.
• Keqin Li is also with the Department of Computer Science, State Univer- Moreover, all decision trees can be trained concurrently,
sity of New York, New Paltz, NY 12561, USA.
hence it is also suitable for parallelization.
 2

1.2 Our Contributions proposed an algorithm for noisy and imbalanced data. Yu
In this paper, we propose a Parallel Random Forest (PRF) et al. [20] and Biau [21] focused on RF for high-dimensional
algorithm for big data that is implemented on the Apache and noisy data and applied RF in many applications such
Spark platform. The PRF algorithm is optimized based on a as multi-class action detection and facial feature detection,
hybrid approach combining data-parallel and task-parallel and achieved a good effort. Based on the existing research
optimization. To improve the classification accuracy of PRF, results, we propose a new optimization approach in this
an optimization is proposed prior to the parallel process. paper to address the problem of high-dimensional and noisy
Extensive experiment results indicate the superiority of PRF data, which reduces the dimensionality of the data accord-
and depict its significant advantages over other algorithms ing to the structure of the RF and improves the algorithm’s
in terms of the classification accuracy and performance. Our accuracy with a low computational cost.
contributions in this paper are summarized as follows. Focusing on the performance of classification algorithms
for large-scale data, numerous studies on the intersection
• An optimization approach is proposed to improve of parallel/distributed computing and the learning of tree
the accuracy of PRF, which includes a dimension- models were proposed. Basilico et al. [22] proposed a
reduction approach in the training process and a COMET algorithm based on MapReduce, in which multiple
weighted voting method in the prediction process. RF ensembles are built on distributed blocks of data. Svore
• A hybrid parallel approach of PRF is utilized to et al. [23] proposed a boosted decision tree ranking algo-
improve the performance of the algorithm, com- rithm, which addresses the speed and memory constraints
bining data-parallel and task-parallel optimization. by distributed computing. Panda et al. [24] introduced a
In the data-parallel optimization, a vertical data- scalable distributed framework based on MapReduce for
partitioning method and a data-multiplexing method the parallel learning of tree models over large datasets. A
are performed. parallel boosted regression tree algorithm was proposed in
• Based on the data-parallel optimization, a task- [25] for web search ranking, in which a novel method for
parallel optimization is proposed and implemented parallelizing the training of GBRT was performed based on
on Spark. A training task DAG of PRF is constructed data partitioning and distributed computing.
based on the RDD model, and different task sched- Focusing on resource allocation and task-parallel exe-
ulers are invoked to perform the tasks in the DAG. cution in a parallel and distributed environment, Warneke
The performance of PRF is improved noticeably. et al. [26] implemented a dynamic resource allocation for
The rest of the paper is organized as follows. Section 2 efficient parallel data processing in a cloud environment.
reviews the related work. Section 3 gives the RF algorithm Lena et al. [27] carried out an energy-aware scheduling of
optimization from two aspects. The parallel implementation MapReduce jobs for big data applications. Luis et al. [28]
of the RF algorithm on Spark is developed in Section 4. proposed a robust resource allocation of data processing on
Experimental results and evaluations are shown in Section 5 a heterogeneous parallel system, in which the arrival time of
with respect to the classification accuracy and performance. datasets are uncertainty. Zhang et al. [29] proposed an evo-
Finally, Section 6 presents a conclusion and future work. lutionary scheduling of dynamic multitasking workloads
for big data analysis in an elastic cloud. Meanwhile, our
team also focused on parallel tasks scheduling on heteroge-
2 R ELATED W ORK neous cluster and distributed systems and achieved positive
Although traditional data processing techniques have results[30, 31].
achieved good performance for small-scale and low- Apache Spark Mllib [32] parallelized the RF algorithm
dimensional datasets, they are difficult to be applied to (referred to Spark-MLRF in this paper) based on a data-
large-scale data efficiently [10–12]. When a dataset becomes parallel optimization to improve the performance of the
more complex with characteristics of a complex structure, algorithm. However, there exist many drawbacks in the
high dimensionality, and a large size, the accuracy and Spark-MLRF. First, in the stage of determining the best split
performance of traditional data mining algorithms are sig- segment for continuous features, a method of sampling for
nificantly declined [13]. each partition of the dataset is used to reduce the data trans-
Due to the need to address the high-dimensional and mission operations. However, the cost of this method is its
noisy data, various improvement methods have been in- reduced accuracy. In addition, because the data-partitioning
troduced by researchers. Xu [14] proposed a dimension- method in Spark-MLRF is a horizontal partition, the data
reduction method for the registration of high-dimensional communication of the feature variable gain ratio computing
data. The method combines datasets to obtain an image is a global communication.
pair with a detailed texture and results in improved image To improve the performance of the RF algorithm and
registration. Tao et al. [15] and Lin et al. [16] introduced mitigate the data communication cost and workload imbal-
some classification algorithms for high-dimensional data to ance problems of large-scale data in parallel and distributed
address the issue of dimension-reduction. These algorithm- environments, we propose a hybrid parallel approach for
s use multiple kernel learning framework and multilevel RF combining data-parallel and task-parallel optimization
maximum margin features and achieve efficient dimension- based on the Spark RDD and DAG models. In compari-
ality reduction in binary classification problems. Strobl [17] son with the existing study results, our method reduces
and Bernard [18] studied the variable importance measures the volume of the training dataset without decreasing the
of RF and proposed some improved models for it. Taghi et algorithm’s accuracy. Moreover, our method mitigates the
al. [19] compared the boosting and bagging techniques and data communication and workload imbalance problems of
 3

large-scale data in parallel and distributed environments. SOOB = {OOB1 , OOB2 , ..., OOBk },
∩ ∪
where k ≪ N , Si OOBi = ϕ and Si OOBi = S . To
3 R ANDOM F OREST A LGORITHM O PTIMIZATION obtain the classification accuracy of each tree model, these
Owing to the improvement of the classification accuracy for OOB sets are used as testing sets after the training process.
high-dimensional and large-scale data, we propose an opti- Step 2. Constructing each decision tree model.
mization approach for the RF algorithm. First, a dimension- In an RF model, each meta decision tree is created by
reduction method is performed in the training process. a C4.5 or CART algorithm from each training subset Si .
Second, a weighted voting method is constructed in the pre- In the growth process of each tree, m feature variables of
diction process. After these optimizations, the classification dataset Si are randomly selected from M variables. In each
accuracy of the algorithm is evidently improved. tree node’s splitting process, the gain ratio of each feature
variable is calculated, and the best one is chosen as the
splitting node. This splitting process is repeated until a leaf
3.1 Random Forest Algorithm node is generated. Finally, k decision trees are trained from
The random forest algorithm is an ensemble classifier al- k training subsets in the same way.
gorithm based on the decision tree model. It generates k Step 3. Collecting k trees into an RF model.
different training data subsets from an original dataset using The k trained trees are collected into an RF model, which
a bootstrap sampling approach, and then, k decision trees is defined in Eq. (1):
are built by training these subsets. A random forest is finally
constructed from these decision trees. Each sample of the ∑
k

testing dataset is predicted by all decision trees, and the H(X, Θj) = hi (x, Θj), (j = 1, 2, ..., m), (1)
i=1
final classification result is returned depending on the votes
of these trees. where hi (x, Θj) is a meta decision tree classifier, X are the
The original training dataset is formalized as S = input feature vectors of the training dataset, and Θj is an
{(xi , yj ), i = 1, 2, ..., N ; j = 1, 2, ..., M }, where x is a independent and identically distributed random vector that
sample and y is a feature variable of S . Namely, the orig- determines the growth process of the tree.
inal training dataset contains N samples, and there are M
feature variables in each sample. The main process of the 3.2 Dimension Reduction for High-Dimensional Data
construction of the RF algorithm is presented in Fig. 1. To improve the accuracy of the RF algorithm for the high-
dimensional data, we present a new dimension-reduction
method to reduce the number of dimensions according
to the importance of the feature variables. In the training
process of each decision tree, the Gain Ratio (GR) of each
feature variable of the training subset is calculated and
sorted in descending order. The top k variables (k ≪ M )
in the ordered list are selected as the principal variables,
and then, we randomly select (m − k) further variables
from the remaining (M − k) ones. Therefore, the number
of dimensions of the dataset is reduced from M to m. The
process of dimension-reduction is presented in Fig. 2.

Fig. 1. Process of the construction of the RF algorithm

The steps of the construction of the random forest algo-

rithm are as follows.
Step 1. Sampling k training subsets.
In this step, k training subsets are sampled from the
original training dataset S in a bootstrap sampling man-
ner. Namely, N records are selected from S by a random
sampling and replacement method in each sampling time. Fig. 2. Dimension-reduction in the training process
After the current step, k training subsets are constructed as
a collection of training subsets ST rain : First, in the training process of each decision tree, the
entropy of each feature variable is calculated prior to the
ST rain = {S1 , S2 , ..., Sk }. node-splitting process. The entropy of the target variable in
At the same time, the records that are not to be selected the training subset Si (i = 1, 2, ..., k ) is defined in Eq. (2):
in each sampling period are composed as an Out-Of-Bag
(OOB) dataset. In this way, k OOB sets are constructed as a
∑
c1
Entropy(Si ) = −pa log pa , (2)
collection of SOOB :
a=1
 4

where c1 is the number of different values of the target

variable in Si , and pa is the probability of the type of value GR(yij )
V I(yij ) = ∑M . (7)
a within all types in the target variable subset. (a=1) GR(y(i,a) )
Second, the entropy for each input variable of Si , except
The importance values of all feature variables are sorted
for the target variable, is calculated. The entropy of each
in descending order, and the top k (k ≪ M, k < m) values
input variable yij is defined in Eq. (3):
are selected as the most important. We then randomly select
(m − k) further feature variables from the remaining (M −
∑ |S(v,i) | k) ones. Thus, the number of dimensions of the dataset is
Entropy(yij ) = Entropy(v(yij )), (3)
|Si | reduced from M to m. Taking the training subset Si as an
v∈V (yij )
example, the detailed steps of the dimension-reduction in
where the elements of Eq. (3) are described in Table 1. the training process are presented in Algorithm 3.1.

Algorithm 3.1 Dimension-reduction in the training process

TABLE 1
Explanation of the elements of Eq. (3) Input:
Si : the ith training subset;
Element Description k: the number of important variables selected by V I ;
yij the j -th feature variable of Si , j = 1, 2, ..., M . m: the number of the selected feature variables.
V (yij ) the set of all possible values of yij . Output:
|Si | the number of samples in Si . Fi : a set of m important feature variables of Si .
1: create an empty string array Fi ;
S(v,i) a sample subset in Si , where the value of yj is v .
2: calculate Entropy(Si ) for the target feature variable;
|S(v,i) | the number of the sample subset S(v,i) .
3: for each feature variable yij in Si do
4: calculate Entropy(yij ) for each input feature variable;
Third, the self-split information I(yij ) of each input 5: calculate gain G(yij ) ← Entropy(Si ) − Entropy(yij );
variable is calculated, as defined in Eq. (4): 6: calculate
∑c2 split information I(yij ) ←
a=1 −p(a,j) log2 (p(a,j) );
∑
c2 7:
G(yij )
calculate gain ratio GR(yij ) ← I(yij )
;
I(yij ) = −p(a,j) log2 (p(a,j) ), (4) 8: end for
GR(yij )
a=1 9: calculate variable importance V I(yij ) ← ∑M GR(y )
(a=1) (i,a)
where c2 is the number of different values of yij , and p(a,j) for feature variable yij ;
is the probability of the type of value a within all types 10: sort M feature variables in descending order by V I(yij );
in variable yj . Then, the information gain of each feature 11: put top k feature variables to Fi [0, ..., k − 1];
12: set c ← 0;
variable is calculated, as defined in Eq. (5): 13: for j = k to M − 1 do
14: while c < (m − k) do
G(yij ) =Entropy(Si ) − Entropy(yij ) 15: select yij from (M − k) randomly;
16: put yij to Fi [k + c];
=Entropy(Si ) 17: c ← c + 1;
∑ |S(v,i) | (5) 18: end while
− Entropy(v(yij )), 19: end for
|Si |
v∈V (yij ) 20: return Fi .

where v(yj ) ∈ V (yj ). In comparison with the original RF algorithm, our

By using the information gain to measure the feature dimension-reduction method ensures that the m selected
variables, the largest value is selected easily, but it will lead feature variables are optimal while maintaining the same
to an over fitting problem. To overcome this problem, a gain computational complexity as the original algorithm. This
ratio value is taken to measure the feature variables, and balances the accuracy and diversity of the feature selection
the features with the maximum value are selected. The gain ensemble of the RF algorithm and prevents the problem of
ratio of the feature variable yij is defined in Eq. (6): classification over fitting.
G(yij )
GR(yij ) = . (6) 3.3 Weighted Voting Method
I(yij )
In the prediction and voting process, the original RF algo-
To reduce the dimensions of the training dataset, we rithm uses a traditional direct voting method. In this case, if
calculate the importance of each feature variable according the RF model contains noisy decision trees, it likely leads to
to the gain ratio of the variable. Then, we select the most a classification or regression error for the testing dataset.
important features and delete the ones with less importance. Consequently, its accuracy is decreased. To address this
The importance of each feature variable is defined as fol- problem, a weighted voting approach is proposed in this
lows. section to improve the classification accuracy for the testing
Definition 1. The importance of each feature variable in dataset. The accuracy of the classification or regression of
a training subset refers to the portion of the gain ratio of the each tree is regarded as the voting weight of the tree.
feature variable compared with the total feature variables. After the training process, each OOB set OOBi is tested
The importance of feature variable yij is defined as V I(yij ) by its corresponding trained tree hi . Then, the classification
in Eq. (7): accuracy CAi of each decision tree hi is computed.
 5

Definition 2. The classification accuracy of a decision the testing data. Hence, this improves the overall classifica-
tree is defined as the ratio of the average number of votes tion accuracy of RF and reduces the generalization error.
in the correct classes to that in all classes, including error
classes, as classified by the trained decision tree. The classi- 3.4 Computational Complexity
fication accuracy is defined in Eq. (8):
The computational complexity of the original RF algorithm
I(hi (x) = y) is O(kM N log N ), where k is the number of decision trees
CAi = ∑ , (8) in RF, M is the number of features, N is the number of
I(hi (x) = y) + I(hi (x) = z)
samples, and log N is the average depth of all tree models.
where I(·) is an indicator function, y is a value in the correct In our improved PRF algorithm with dimension-reduction
class, and z is a value in the error class (z ̸= y ). (PRF-DR) described in Section 3, the time complexity of the
In the prediction process, each record of the testing dimension reduction is O(M N ). The computational com-
dataset X is predicted by all decision trees in the RF model, plexity of the splitting process for each tree node is set as one
and then, a final vote result is obtained for the testing record. unit (1), which contains functions such as entropy(), gain(),
When the target variable of X is quantitative data, the RF is and gainratio() for each feature subspace. After the dimen-
trained as a regression model. The result of the prediction is sion reduction, the number of features is reduced from M
the average value of k trees. The weighted regression result to m (m ≪ M ). Therefore, the computational complexity of
Hr (X) of X is defined in Eq. (9): training a meta tree classifier is O(M N + mN log N ), and
the total computational complexity of the PRF-DR algorithm
1∑
k
is O(k(M N + mN log N )).
Hr (X) = [wi × hi (x)]
k i=1
(9)
1∑
k 4 PARALLELIZATION OF THE R ANDOM F OREST
= [CAi × hi (x)], A LGORITHM ON S PARK
k i=1
To improve the performance of the RF algorithm and miti-
where wi is the voting weight of the decision tree hi . gate the data communication cost and workload imbalance
Similarly, when the target feature of X is qualitative problems of large-scale data in a parallel and distributed en-
data, the RF is trained as a classification model. The result of vironment, we propose a Parallel Random Forest (PRF) algo-
the prediction is the majority vote of the classification results rithm on Spark. The PRF algorithm is optimized based on a
of k trees. The weighted classification result Hc (X) of X is hybrid parallel approach combining data-parallel and task-
defined in Eq. (10): parallel optimization. From the perspective of data-parallel
optimization, a vertical data-partitioning method and a
k
Hc (X) = M ajorityi=1 [wi × hi (x)] data-multiplexing method are performed. These methods
(10)
k
= M ajorityi=1 [CAi × hi (x)]. reduce the volume of data and the number of data trans-
mission operations in the distributed environment without
The steps of the weighted voting method in the predic- reducing the accuracy of the algorithm. From the perspec-
tion process are described in Algorithm 3.2. tive of task-parallel optimization, a dual-parallel approach
is carried out in the training process of the PRF algorithm,
Algorithm 3.2 Weighted voting in the prediction process and a task DAG is created according to the dependence
Input: of the RDD objects. Then, different task schedulers are
X : a testing dataset; invoked to perform the tasks in the DAG. The dual-parallel
P RFtrained : the trained PRF model.
training approach maximizes the parallelization of PRF and
Output:
H(X): the final prediction result for X . improves the performance of PRF. Then task schedulers
1: for each testing data x in X do further minimize the data communication cost among the
2: for each decision tree Ti in P RFtrained do Spark cluster and achieve a better workload balance.
3: predict the classification or regression result hi (x) by
Ti ;
4: end for 4.1 Data-Parallel Optimization
5: end for We introduce the data-parallel optimization of the PRF
6: set classification accuracy CAi as the weight wi of Ti ; algorithm, which includes a vertical data-partitioning and
7: for each testing data x in X do a data-multiplexing approach. First, taking advantage of
8: if (operation type == classification) then
the RF algorithm’s natural independence of feature vari-
9: vote the final result Hc (x) ← M ajorityi=1
k
[wi × hi (x)];
10: H(X) ← Hc (x); ables and the resource requirements of computing tasks,
else if (operation type == regression) a vertical data-partitioning method is proposed for the
11:
∑ then
12: vote the final result Hr (x) ← k1 ki=1 [wi × hi (x)]; training dataset. The training dataset is split into several
13: H(X) ← Hr (x); feature subsets, and each feature subset is allocated to the
14: end if Spark cluster in a static data allocation way. Second, to
15: end for address the problem that the data volume increases linearly
16: return H(X). with the increase in the scale of RF, we present a data-
multiplexing method for PRF by modifying the traditional
In the weighted voting method of RF, each tree classifier sampling method. Notably, our data-parallel optimization
corresponds to a specified reasonable weight for voting on method reduces the volume of data and the number of data
 6

transmission operations without reducing the accuracy of tasks in the training process of each decision tree load the
the algorithm. The increase in the scale of the PRF does not corresponding data from the same feature subset via the DSI
lead to a change in the data size and storage location. table. Thus, each feature subset is reused effectively, and the
volume of the training dataset will not increase any more
4.1.1 Vertical Data Partitioning despite the expansion of the RF scale.
In the training process of PRF, the gain-ratio computing First, we create a DSI table to save the data indexes
tasks of each feature variable take up most of the training generated in all sampling times. As mentioned in Section
time. However, these tasks only require the data of the 3.1, the scale of a RF model is k . Namely, there are k
current feature variable and the target feature variable. sampling times for the training dataset, and N data indexes
Therefore, to reduce the volume of data and the data com- are noted down in each sampling time. An example of the
munication cost in a distributed environment, we propose DSI table of PRF is presented in Table 2.
a vertical data-partitioning approach for PRF according to
the independence of feature variables and the resource TABLE 2
Example of the DSI table of PRF
requirements of computing tasks. The training dataset is
divided into several feature subsets. Data indexes of training dataset
Assume that the size of training dataset S is N and there Sample0 1 3 8 5 10 0 ...

Sampling times
are M feature variables in each record. y0 ∼ yM −2 are Sample1 2 4 1 9 7 8 ...
the input feature variables, and yM −1 is the target feature Sample2 9 1 12 92 2 5 ...
variable. Each input feature variable yj and the variable Sample3 3 8 87 62 0 2 ...
yM −1 of all records are selected and generated to a feature ... ... ... ... ... ... ... ...
subset F Sj , which is represented as: Samplek−1 9 1 4 43 3 5 ...
 
< 0, y0j , y0(M −1) >, Second, the DSI table is allocated to all slave nodes of
 < 1, y1j , y1(M −1) >, 
  the Spark cluster together with all feature subsets. In the
 ...,  subsequent training process, the gain-ratio computing tasks
F Sj =  ,
 < i, yij , yi(M −1) >,  of different trees for the same feature variable are dispatched
 ..., 
to the slaves where the required subset is located.
< (N − 1), y(N −1)j , y(N −1)(M −1) > Third, each gain-ratio computing task accesses the rele-
where i is the index of each record of the training dataset S , vant data indexes from the DSI table, and obtains the feature
and j is the index of the current feature variable. In such a records from the same feature subset according to these
way, S is split into (M −1) feature subsets before dimension- indexes. The process of the data-multiplexing method of
reduction. Each subset is loaded as an RDD object and is PRF is presented in Fig. 4.
independent of the other subsets. The process of the vertical
data-partitioning is presented in Fig. 3.

Fig. 4. Process of the data-multiplexing method of PRF

Fig. 3. Process of the vertical data-partitioning method In Fig. 4, each RDDF S refers to an RDD objec-
t of a feature subset, and each TGR refers to a gain-
ratio computing task. For example, we allocate tasks
4.1.2 Data-Multiplexing Method {TGR1.1 , TGR1.2 , TGR1.3 } to Slave1 for the feature subset
To address the problem that the volume of the sampled RDDF S1 , allocate tasks {TGR2.1 , TGR2.2 , TGR2.3 } to Slave2
training dataset increases linearly with the increase in the for RDDF S2 , and allocate tasks {TGR3.1 , TGR3.2 , TGR3.3 } to
RF scale, we present a data-multiplexing method for PRF Slave3 for RDDF S3 . From the perspective of the decision
by modifying the traditional sampling method. In each trees, tasks in the same slave node belong to different
sampling period, we do not copy the sampled data but trees. For example, tasks TGR1.1 , TGR1.2 , and TGR1.3 in the
just note down their indexes into a Data-Sampling-Index Slave1 belong to “DecisionT ree1”, “DecisionT ree2”, and
(DSI) table. Then, the DSI table is allocated to all slave “DecisionT ree3”, respectively. These tasks obtain records
nodes together with the feature subsets. The computing from the same feature subset according to the corresponding
 7

indexes in DSI, and compute the gain ratio of the feature Algorithm 4.1 Vertical data-partitioning and static data
variable for different decision trees. After that, the interme- allocation of PRF
diate results of these tasks are submitted to the correspond- Input:
ing subsequent tasks to build meta decision trees. Results RDDo : an RDD object of the original training dataset S .
of the tasks {TGR1.1 , TGR2.1 , TGR3.1 } are combined for the Output:
tree node splitting process of “DecisionT ree1”, and results LF S : a list of the indexes of each feature subset’s RDD
of the tasks {TGR1.2 , TGR2.2 , TGR3.2 } are combined for that object and the allocated slave nodes.
1: for j = 0 to (M − 2) do
of “DecisionT ree2”.
2: RDDF Sj ← RDDo .map
3: < i, yij , yi(M −1) >← RDDo .verticalPartition(j);
4.1.3 Static Data Allocation 4: end map.collect()
5: slaves ← findAvailableSlaves().sortbyIP();
To achieve a better balance of data storage and workload, 6: if RDDF Sj .size < slaves[0].availablesize then
after the vertical data-partitioning, a static data allocation 7: dataAllocation(RDDF Sj , slaves[0]);
method is applied for the feature subsets. Namely, these 8: slaves[0].availablesize ← slaves[0].availablesize -
subsets are allocated to a distributed Spark cluster before RDDF Sj .size;
the training process of PRF. Moreover, because of the dif- 9: LF S ← < RDDF Sj .id, slaves[0].nodeid >;
ference of the data type and volume of each feature subset, 10: else
11: while RDDF Sj ̸= null do
the workloads of the relevant subsequent computing tasks ′ ′′
12: (RDDF Sj , RDDF Sj ) ← dataPartition(RDDF Sj ,
will be different as well. As we know, a Spark cluster is slaves[i].availablesize));
constructed by a master node and several slave nodes. We 13:
′
dataAllocation(RDDF Sj , slaves[i]);
define our allocation function to determine each feature ′
14: RDDF Sj .persist();
subset be allocated to which nodes, and allocate each feature 15: slaves[i].availablesize ← slaves[i].availablesize -
′
subset according to its volume. There are 3 scenarios in the RDDF Sj .size;
data allocation scheme. Examples of the 3 scenarios of the 16: slavesids ← slaves[i].nodeid;
′′
data allocation method are shown in Fig. 5. 17: RDDF Sj ← RDDF Sj ;
18: i ← i + 1;
19: end while
20: LF S ← < RDDF Sj .id, slavesids >;
21: end if
22: end for
23: return LF S .

4.2 Task-Parallel Optimization

Each decision tree of PRF is built independent of each other,
and each sub-node of a decision tree is also split indepen-
Fig. 5. Example of 3 scenarios of the data allocation
dently. The structures of the PRF model and decision tree
model make the computing tasks have natural parallelism.
In Fig. 5, (a) when the size of a feature subset is greater
Based on the results of the data-parallel optimization, we
than the available storage capacity of a slave node, this sub-
propose a task-parallel optimization for PRF and implement
set is allocated to limited multiple slaves that have similar
it on Spark. A dual-parallel approach is carried out in the
physical locations. (b) When the size of a feature subset is
training process of PRF, and a task DAG is created according
equal to the available storage capacity of a slave node, the
to the dual-parallel training process and the dependence of
subset is allocated to the node. (c) When the size of a feature
the RDD objects. Then, different task schedulers are invoked
subset is smaller than the available storage capacity of a
to perform the tasks in the DAG.
slave node, this node will accommodate multiple feature
subsets. In case (a), the data communication operations of
the subsequent gain-ratio computing tasks occur among the 4.2.1 Parallel Training Process of PRF
slave nodes where the current feature subset is located. In our task-parallel optimization approach, a dual-parallel
These data operations are local communications but not training approach is proposed in the training process of
global communications. In cases (b) and (c), no data com- PRF on Spark. k decision trees of the PRF model are built
munication operations occur among different slave nodes in parallel at the first level of the training process. And
in the subsequent gain-ratio computation process. The steps (M −1) feature variables in each decision tree are calculated
of the vertical data-partitioning and static data allocation of concurrently for tree node splitting at the second level of the
PRF are presented in Algorithm 4.1. training process.
In Algorithm 4.1, RDDo is split into (M − 1) RDDF S There are several computing tasks in the training process
objects via the vertical data-partitioning function firstly. of each decision tree of PRF. According to the required data
Then, each RDDF S is allocated to slave nodes according resources and the data communication cost, the computing
to its volume and the available storage capacity of the slave tasks are divided into two types, gain-ratio computing tasks
nodes. To reuse the training dataset, each RDD object of the and node-splitting tasks, which are defined as follows.
feature subset is allocated and persisted to Spark cluster via Definition 3. Gain-ratio-computing task (TGR ) is a task
a dataAllocation() function and a persist() function. set that is employed to compute the gain ratio of a feature
 8

variable from the corresponding feature subset, which in-

cludes a series of calculations for each feature variable, i.e.,
the entropy, the self-split information, the information gain,
and the gain ratio. The results of TGR tasks are submitted to
the corresponding subsequent node-splitting tasks.
Definition 4. Node-splitting task (TN S ) is a task set that
is employed to collect the results of the relevant TGR tasks
and split the decision tree nodes, which includes a series
of calculations for each tree node, such as determining the
best splitting variable holds the highest gain ratio value and
splitting the tree node by the variable. After the tree node
splitting, the results of TN S tasks are distributed to each
slave to begin the next stage of the PRF’s training process.
The steps of the parallel training process of the PRF
model are presented in Algorithm 4.2.

Algorithm 4.2 Parallel training process of the PRF model Fig. 6. Example of the task DAG for a decision tree of PRF

Input:
k: the number of decision trees of the PRF model; splitting variable and the data index list of {v01 , v02 , v03 }.
TDSI : the DSI table of PRF;
In stage 2, because y0 is the splitting feature, there
LF S : a list of the indexes of each feature subset’s RDD
object and the allocated slave nodes. is no TGR task for F S0 . The potential workload balance
Output: problem of this issue will be discussed in Section 4.3.4.
P RFtrained : the trained PRF model. New TGR tasks are generated for all other feature subsets
1: for i = 0 to (k − 1) do according to the result of TN S1 . Due to the data index list
2: for j = 0 to (M − 2) do of {v01 , v02 , v03 }, there are no more than 3 TGR tasks for
3: load feature subset RDDF Sj ← loadData(LF S [i]); each feature subset. For example, tasks TGR2.11 , TGR2.12 ,
//TGR :
4: RDD(GR,best) ← sc.parallelize(RDDF Sj ).map
and TGR2.13 calculate the data of F S1 with the indexes
5: load sampled data RDD(i,j) ← (TDSI [i], RDDF Sj ); corresponding to v01 , v02 , and v0 3, respectively. And the
6: calculate the gain ratio GR(i,j) ← GR(RDD(i,j) ); condition is similar in tasks for F S2 ∼ F S(m−2) . Then, the
7: end map results of tasks {TGR2.11 , TGR2.21 , TGR2.(m−2)1 } are submit-
//TN S : ted to task TN S2.1 for the same sub-tree-node splitting. Tasks
8: RDD(GR,best) .collect().sorByKey(GR).top(1); of other tree nodes and other stages are performed similarly.
9: for each value y(j,v) in RDD(GR,best) do
In such a way, a task DAG of the training process of each
10: split tree node N odej ←< y(j,v) , V alue >;
11: append N odej to Ti ; decision tree model is built. In addition, k DAGs are built
12: end for respectively for the k decision trees of the PRF model.
13: end for
14: P RFtrained ← Ti ; 4.2.2 Task-Parallel Scheduling
15: end for After the construction of the task DAGs of all the decision
16: return P RFtrained . trees, the tasks in these DAGs are submitted to the Spark
task scheduler. There exist two types of computing tasks
According to the parallel training process of PRF and the in the DAG, which have different resource requirements
dependence of each RDD object, each job of the program of and parallelizables. To improve the performance of PRF
PRF’s training process is split into different stages, and a efficiently and further minimize the data communication
task DAG is constructed with the dependence of these job cost of tasks in the distributed environment, we invoke two
stages. Taking a decision tree model of PRF as an example, different task-parallel schedulers to perform these tasks.
a task DAG of the training process is presented in Fig. 6. In Spark, the T askSchedulerListener module moni-
There are several stages in the task DAG, which corre- tors the submitted jobs, splits the job into different stages
spond to the levels of the decision tree model. In stage 1, and tasks, and submits these tasks to the T askScheduler
after the dimension-reduction, (m − 1) TGR tasks (TGR1.0 module. The T askScheduler module receives the tasks and
∼ TGR1.(m−2) ) are generated for the (m − 1) input feature allocates and executes them using the appropriate executors.
variables. These TGR s compute the gain ratio the corre- According to the different allocations, the T askScheduler
sponding feature variable, and submit their results to TN S1 . module includes 3 sub-modules, such as LocalScheduler,
TN S1 finds the best splitting variable and splits the first tree ClusterScheduler, and M essosScheduler. Meanwhile,
node for the current decision tree model. Assuming that y0 each task holds 5 types of locality property val-
is the best splitting variable at the current stage, and the ue: P ROCESS LOCAL, N ODE LOCAL, N O P REF ,
value of y0 is in the range of {v01 , v02 , v03 }. Hence, the P ACK LOCAL, and AN Y . We set the value of the locality
first tree node is constructed by y0 , and 3 sub-nodes are properties of these two types of tasks and submit them into
split from the node, as shown in Fig. 6(b). After tree node different task schedulers. We invoke LocalScheduler for
splitting, the intermediate result of TN S1 are distributed TGR tasks and ClusterScheduler to perform TN S tasks.
to all slave nodes. The result includes information of the (1) LocalScheduler for TGR tasks.
 9

The LocalScheduler module is a thread pool of the

local computer, all tasks submitted by DAGScheduler is
executed in the thread pool, and the results will then be
returned to DAGScheduler. We set the locality property
value of each TGR as N ODE LOCAL and submit it to
a LocalScheduler module. In LocalScheduler, all TGR
tasks of PRF are allocated to the slave nodes where the
corresponding feature subsets are located. These tasks are
independent of each other, and there is no synchronization
restraint among them. If a feature subset is allocated to
multiple slave nodes, the corresponding TGR tasks of each
decision tree are allocated to these nodes. And there exist
local data communication operations of the tasks among
these nodes. If one or more feature subsets are allocated
to one slave node, the corresponding TGR tasks are posted
to the current node. And there is no data communication
operation between the current node and the others in the
subsequent computation process.
(2) ClusterScheduler for TN S tasks.
The ClusterScheduler module monitors the execution
situation of the computing resources and tasks in the whole
Spark cluster and allocates tasks to suitable workers. As
mentioned above, TN S tasks are used to collect the results
of the corresponding TGR tasks and split the decision tree Fig. 7. Task-parallel scheduling based on the DAG in Fig. 6
nodes. TN S tasks are independent of all feature subsets
and can be scheduled and allocated in the whole Spark
cluster. In addition, these TN S tasks rely on the results 4.3.1 Computational Complexity Analysis
of the corresponding TGR tasks, therefore, there is a wait As discussed in Section 3.4, the total computational com-
and synchronization restraint for these tasks. Therefore, we plexity of the improved PRF algorithm with dimension-
invoke the ClusterScheduler to perform TN S tasks. We reduction is O(k(M N + mN log N )). After the paralleliza-
set the locality property value of each TN S as AN Y and tion of the PRF algorithm on Spark, M features of train-
submit to a ClusterScheduler module. The task-parallel ing dataset are calculated in parallel in the process of
scheduling scheme for TN S tasks is described in Algorithm dimension-reduction, and k trees are trained concurrently.
4.3. A diagram of task-parallel scheduling for the tasks in Therefore, the theoretical computational complexity of the
k(M N +mN log N )
the above DAG is shown in Fig. 7. PRF algorithm is O( k×M ) ≈ O(N (log N + 1)).

Algorithm 4.3 Task-parallel scheduling for TN S tasks 4.3.2 Data Volume Analysis
Input: Taking advantage of the data-multiplexing method, the
T SN S : a task set of all TN S submitted by DAGScheduler . training dataset is reused effectively. Assume that the vol-
Output: ume of the original dataset is (N × M ) and the RF model’s
ERT S : the execution results of T SN S . scale is k , the volumes of the sampled training dataset in
1: create manager ← new TaskSetManager(T SN S );
the original RF and Spark-MLRF are both (N × M × k).
2: append to taskset manager activeT SQueue ← manager ;
3: if hasReceivedTask == false then In our PRF, the volume of the sampled training dataset
4: create starvationT imer ← scheduleAtFixedRate(new is (N × 2 × (M − 1)) ≈ (2N M ). Moreover, the increase
TimerTask); of the scale of PRF does not lead to changes in the data
5: rank the priority of T S2 ← activeT SQueue.FIFO(); size and storage location. Therefore, compared with the
6: for each task Ti in T S2 do sampling method of the original RF and Spark-MLRF, the
7: get available worker executora from workers; data-parallel method of our PRF decreases the total volume
8: ERT S ← executora .launchTask(Ti .taskid);
9: end for of the training dataset for PRF.
10: end if
11: return ERT S . 4.3.3 Data Communication Analysis
In PRF, there exist data communication operations in the
process of data allocation and the training process. Assume
that there are n slaves in a Spark cluster, and the data
4.3 Parallel Optimization Method Analysis volume of the sampled training dataset is (2N M ). In the
We discuss our hybrid parallel optimization method from process of data allocation, the average data communication
5 aspects as follows. In comparison with Spark-MLRF and cost is ( 2MnN ). In the process of the PRF model training, if a
other parallel methods of RF, our hybrid parallel opti- feature subset is allocated to several computer nodes, local
mization approach of PRF achieves advantages in terms of data communication operations of the subsequent comput-
performance, workload balance, and scalability. ing tasks occur among these nodes. If one or more feature
 10

subsets are allocated to one computer node, there is no nodes. Each node executes in Ubuntu 12.04.4 and has one
data communication operation among different nodes in the Pentium (R) Dual-Core 3.20GHz CPU and 8GB memory.
subsequent computation process. Generally, there is a small All nodes are connected by a high-speed Gigabit network
amount of data communication cost for the intermediate and are configured with Hadoop 2.5.0 and Spark 1.1.0. The
results in each stage of the decision tree’s training pro- algorithm is implemented in Scala 2.10.4. Two groups of
cess. The vertical data-partitioning and static data allocation datasets with large scale and high dimensionality are used
method mitigates the amount of data communication in the in the experiments. One is from the UCI machine learning
distributed environment and overcomes the performance repository [33], as shown in Table 3. Another is from a actual
bottleneck of the traditional parallel method. medical project, as shown in Table 4.
4.3.4 Resource and Workload Balance Analysis
TABLE 3
From the view point of the entire training process of PRF Datasets from the UCI machine learning repository
in the whole Spark cluster, our hybrid parallel optimization Datasets Instances Features Classes Data Size Data Size
approach achieves a better storage and workload balance (Original) (Maximum)
than other algorithms. One reason is that because the d- URL Reputation 2396130 3231961 5 2.1GB 1.0TB
ifferent volumes of feature subsets might lead to different (URL)
workloads of the TGR tasks for each feature variable, we You Tube Video 120000 1000000 14 25.1GB 2.0TB
allocate the feature subsets to the Spark cluster according Games (Games)
to its volume. A feature subset with a large volume is Bag of Words 8000000 100000 24 15.8GB 1.3TB
allocated to multiple slave nodes. And the corresponding (Words)
TGR tasks are scheduled among these nodes in parallel. A Gas sensor arrays 1800000 1950000 15 50.2GB 2.0TB
(Gas)
feature subsets with a small volume are allocated to one
slave node. And the corresponding TGR tasks are scheduled
on the current node.
TABLE 4
A second reason is that with the tree nodes’ splitting, Datasets from a medical project
the slave nodes where the split variables’ feature subsets are
Datasets Instances Features Classes Data size Data size
located will revert to an idle status. From the view point
(Original) (Maximum)
of the entire training process of PRF, profit from the data-
multiplexing method of PRF, each feature subset is shared Patient 279877 25652 18 3.8GB 1.5TB
and reused by all decision trees, and it might be split for Outpatient 3657789 47562 9 10.6GB 1.0TB
different tree nodes in different trees. That is, although a Medicine 7502058 52460 12 20.4GB 2.0TB
feature subset is split and useless to a decision tree, it is Cancer 3568000 46532 21 5.8GB 2.0TB
still useful to other trees. Therefore, our PRF not only does
not lead to the problem of waste of resources and workload In Table 3 and Table 4, Datasize(Original) refers to
imbalance, but also makes full use of the data resources and the original size of the data from the UCI and the project,
achieves an overall workload balance. and Datasize(M aximum) refers to the peak size of data
sampled by all of the comparison algorithms.
4.3.5 Algorithm Scalability Analysis In the Spark platform, the training data not be loaded
We discuss the stability and scalability of our PRF algorithm into the memory as a whole. Spark can be used to process
from 3 perspectives. (1) The data-multiplexing method of datasets that are greater than the total cluster memory ca-
PRF makes the training dataset be reused effectively. When pacity. RDD objects in a single executor process are accessed
the scale of PRF expands, namely, the number of decision by an iteration, and the data are buffered or thrown away
trees increases, the data size and the storage location of the after the processing. The cost of memory is very small when
feature subsets need not change. It only results in an increase there is no requirement of caching the results of the RDD
in computing tasks for new decision trees and a small objects. In this case, the results of the iterations are retained
amount of data communication cost for the intermediate in a memory pool by the cache manager. When the data in
results of these tasks. (2) When the Spark cluster’s scale the memory are not applicable, they will be saved on disk.
expands, only a few feature subsets with a high storage In this case, part of the data can be kept in the memory and
load are migrated to the new computer nodes to achieve the rest is stored in the disk. Therefore, the training data
storage load and workload balance. (3) When the scale of with the peak size of 2.0TB can be executed on Spark.
the training dataset increases, it is only necessary to split
feature subsets from the new data in the same vertical 5.2 Classification Accuracy
data-partitioning way, and append each new subset to the We evaluate the classification accuracy of PRF by compari-
corresponding original one. Therefore, we can draw the son with RF, DRF, and Spark-MLRF.
conclusion that our PRF algorithm with the hybrid parallel
optimization method achieves good stability and scalability. 5.2.1 Classification Accuracy for Different Tree Scales
To illustrate the classification accuracy of PRF, experiments
5 E XPERIMENTS
are performed for the RF, DRF [18], Spark-MLRF, and PRF
5.1 Experiment Setup algorithms. The datasets are outlined in Table 3 and Table 4.
All the experiments are performed on a Spark cloud plat- Each case involves different scales of the decision tree. The
form, which is built of one master node and 100 slave experimental results are presented in Fig. 8.
 11

greater than that of DRF by 8.6%, on average, and 10.7%

higher in the best case when the number of samples is equal
to 3,000,000. The classification accuracy of PRF is greater
than that of Spark-MLRF by 8.1%, on average, and 11.3%
higher in the best case when the number of samples is
equal to 3,000,000. For Spark-MLRF, because of the method
of sampling for each partition of the dataset, as the size
of the dataset increases, the ratio of the random selection
of the dataset increases, and the accuracy of Spark-MLRF
decreases inevitably. Therefore, compared with RF, DRF,
and Spark-MLRF, PRF improves the classification accuracy
significantly for different scales of datasets.

5.2.3 OOB Error Rate for Different Tree Scales

We observe the classification error rate of PRF under dif-
ferent conditions. In each condition, the P atient dataset is
Fig. 8. Average classification accuracy for different tree scales chosen, and two scales (500 and 1000) of decision trees are
constructed. The experimental results are presented in Fig.
10 and Table 5.
Fig. 8 shows that the average classification accuracies of
all of the comparative algorithms are not high when the
number of decision trees is equal to 10. As the number of
decision trees increases, the average classification accuracies
of these algorithms increase gradually and have a tendency
toward a convergence. The classification accuracy of PRF is
higher than that of RF by 8.9%, on average, and 10.6% higher
in the best case when the number of decision trees is equal to
1500. It is higher than that of DRF by 6.1%, on average, and
7.3% higher in the best case when the number of decision
trees is equal to 1300. The classification accuracy of PRF Fig. 10. OOB error rates of PRF for different tree scales
is higher than that of Spark-MLRF by 4.6% on average,
and 5.8% in the best case when the number of decision When the number of decision trees of PRF increases, the
trees is equal to 1500. Therefore, compared with RF, DRF, OOB error rate in each case declines gradually and tends to
and Spark-MLRF, PRF improves the classification accuracy a convergence condition. The average OOB error rate of PRF
significantly. is 0.138 when the number of decision trees is equal to 500,
and it is 0.089 when the number of decision trees is equal to
5.2.2 Classification Accuracy for Different Data Sizes 1000.
Experiments are performed to compare the classification TABLE 5
accuracy of PRF with the RF, DRF, and Spark-MLRF al- OOB error rates of PRF for different tree scales
gorithms. Datasets from the project described in Table 4 Tree=500 Tree=1000
are used in the experiments. The experimental results are Rate OOB Class1 Class2 OOB Class1 Class2
presented in Fig. 9. max 0.207 0.270 0.354 0.151 0.132 0.318
min 0.113 0.051 0.092 0.067 0.010 0.121
mean 0.138 0.094 0.225 0.089 0.056 0.175

5.3 Performance Evaluation

Various experiments are constructed to evaluate the perfor-
mance of PRF by comparison with the RF and Spark-MLRF
algorithms in terms of the execution time, speedup, data
volume, and data communication cost.

5.3.1 Average Execution Time for Different Datasets

Experiments are performed to compare the performance
of PRF with that of RF and Spark-MLRF. Four groups of
Fig. 9. Average classification accuracy for different data sizes training datasets are used in the experiments, such as U RL,
Games, Outpatient, and P atient. In these experiments, the
The classification accuracies of PRF in all of the cases are number of decision trees in each algorithm is both 500, and
greater than that of RF, DRF, and Spark-MLRF obviously the number of Spark slaves is 10. The experimental results
for each scale of data. The classification accuracy of PRF is are presented in Fig. 11.
 12

In Fig. 12, because of the different data sizes and contents

of the training data, the execution times of PRF in each case
are different. When the number of slave nodes increases
from 10 to 50, the average execution times of PRF in all cases
obviously decrease. For example, the average execution time
of PRF decreases from 405.4 to 182.6 seconds in the Gas
case and from 174.8 to 78.3 seconds in the M edicine case.
By comparison, the average execution times of PRF in the
other cases decrease less obviously when the number of
slave nodes increases from 50 to 100. For example, the
average execution time of PRF decreases from 182.4 to 76.0
seconds in the Gas case and from 78.3 to 33.0 seconds in
the M edicine case. This is because when the number of the
Spark slaves greater than that of training dataset’s feature
variables, each feature subset might be allocated to multiple
slaves. In such a case, there are more data communication
operations among these slaves than before, which leads to
more execution time of PRF.
Fig. 11. Average execution time of the algorithms for different datasets
5.3.3 Speedup of PRF in Different Environments
Experiments in a stand-alone environment and a Spark
When the data size is small (e.g., less than 1.0GB), the cloud platform are performed to evaluate the speedup of
execution times of PRF and Spark-MLRF are higher than PRF. Because of the different volume of training datasets,
that of RF. The reason is that there is a fixed time required to the execution times of PRF are not in the same range in
submit the algorithms to the Spark cluster and configure different cases. To observe the comparison of the execution
the programs. When the data size is greater than 1.0GB, time intuitively, a normalization of execution time is taken.
the average execution times of PRF and Spark-MLRF are Let T(i,sa) be the execution time of PRF for dataset Si in
less than that of RF in the four cases. For example, in the the stand-alone environment, and first normalized to 1. The
Outpatient case, when the data size grows from 1.0 to execution time of PRF on Spark is normalized as described
500.0GB, the average execution time of RF increases from in Eq. (11):
19.9 to 517.8 seconds, while that of Spark-MLRF increases { T(i,sa)
from 24.8 to 186.2 seconds, and that of PRF increases from ′ T(i,sa) = 1 Stand − alone,
23.5 to 101.3 seconds. Hence, our PRF algorithm achieves Ti = T(i,Spark) (11)
T(i,sa) Spark.
a faster processing speed than RF and Spark-MLRF. When
the data size increases, the benefit is more noticeable. Taking The speedup of PRF on Spark for Si is defined in Eq.
advantage of the hybrid parallel optimization, PRF achieves (12):
significant strengths over Spark-MLRF and RF in terms of
′
performance. T(i,Spark)
Speedup(i,Spark) = ′ . (12)
T(i,sa)
5.3.2 Average Execution Time for Different Cluster Scales
In this section, the performance of PRF on the Spark plat- The results of the comparative analyses are presented in
form for different scales of slave nodes is considered. The Fig. 13. Taking benefits of the parallel algorithm and cloud
number of slave nodes is gradually increased from 10 to environment, the speedup of PRF on Spark tends to increase
100, and the experiment results are presented in Fig. 12. in each experiment with the number of slave nodes. When
the number of slave nodes is equal to 100, the speedup factor
of PRF in all cases is in the range of 60.0 - 87.3, which is less
than the theoretical value (100). Because there exists data
communication time in a distributed environment and a
fixed time for the application submission and configuration,
it is understandable that the whole speedup of PRF is less
than the theoretical value. Due to the different data volumes
and contents, the speedup of PRF in each case is different.
When the number of slave nodes is less than 50, the
speedup in each case shows a rapid growth trend. For
instance, compared with the stand-alone environment, the
speedup factor of Gas is 65.5 when the number of slave
nodes is equal to 50, and the speedup factor of P atient
is 61.5. However, the speedup in each case shows a slow
growth trend when the number of slave nodes is greater
Fig. 12. Average execution time of PRF for different cluster scales than 50. This is because there are more data allocation, task
 13

Fig. 15. Data communication costs of PRF and Spark-MLRF

Fig. 13. Speedup of PRF in different environments

the algorithms. Taking the P atient case as an example, the

scheduling, and data communication operations required results of the comparison of CDC are presented in Fig. 15.
for PRF. From Fig. 15, it is clear that the CDC of PRF are less than
that of Spark-MLRF in all cases, and the distinction is larger
5.3.4 Data Volume Analysis for Different RF Scales with increasing number of slave nodes. Although Spark-
MLRF also uses the data-parallel method, the horizontal
We analyze the volume of the training data in PRF against
partitioning method for training data makes the computing
RF and Spark-MLRF. Taking the Games case as an example,
tasks have to frequent access data across different slaves.
the volumes of the training data in the different RF scales
As the number of slaves increases from 5 to 50, the CDC of
are shown in Fig. 14.
Spark-MLRF increases from 350.0MB to 2180.0MB. Different
from Spark-MLRF, in PRF, the vertical data-partitioning and
allocation method and the task scheduling method make
the most of the computing tasks (TGR ) access data from the
local slave, reducing the amount of data transmission in the
distributed environment. As the number of slaves increases
from 5 to 50, the CDC of PRF increases from 50.0MB to
320.0MB, which is much lower than that of Spark-MLRF.
Therefore, PRF minimizes the CDC of RF in a distributed
environment. The expansion of the cluster’s scale does not
lead to an obviously increase in CDC . In conclusion, our PRF
achieves a superiority and notable advantages over Spark-
MLRF in terms of stability and scalability.

6 C ONCLUSIONS
Fig. 14. Size of training dataset for different RF scales In this paper, a parallel random forest algorithm has been
proposed for big data. The accuracy of the PRF algorithm
In Fig. 14, due to the use of the same horizontal sampling is optimized through dimension-reduction and the weight-
method, the training data volumes of RF and Spark-MLRF ed vote approach. Then, a hybrid parallel approach of
both show a linear increasing trend with the increasing of PRF combining data-parallel and task-parallel optimization
the RF model scale. Contrary, in PRF, the total volume of is performed and implemented on Apache Spark. Taking
all training feature subsets is 2 times the size of the orig- advantage of the data-parallel optimization, the training
inal training dataset. Making use of the data-multiplexing dataset is reused and the volume of data is reduced signif-
approach of PRF, the training dataset is effectively reused. icantly. Benefitting from the task-parallel optimization, the
When the number of decision trees is larger than 2, despite data transmission cost is effectively reduced and the per-
the expansion of RF scale, the volume of the training data formance of the algorithm is obviously improved. Experi-
will not increases any further. mental results indicate the superiority and notable strengths
of PRF over the other algorithms in terms of classification
5.3.5 Data Communication Cost Analysis accuracy, performance, and scalability. For future work, we
Experiments are performed for different scales of the Spark will focus on the incremental parallel random forest algo-
cluster to compare the Data Communication Cost (CDC ) rithm for data streams in cloud environment, and improve
of PRF with that of Spark-MLRF. The suffer-write size of the data allocation and task scheduling mechanism for the
slave nodes in the Spark cluster is monitored as the CDC of algorithm on a distributed and parallel environment.
 14

ACKNOWLEDGMENT Intelligence, IEEE Transactions on, vol. 33, no. 6, pp. 1147–
The research was partially funded by the Key Program of 1160, June 2011.
[17] C. Strobl, A. Boulesteix, T. Kneib, and T. Augustin, “Con-
National Natural Science Foundation of China (Grant Nos. ditional variable importance for random forests,” BMC
61133005, 61432005), the National Natural Science Founda- Bioinformatics, vol. 9, no. 14, pp. 1–11, 2007.
tion of China (Grant Nos. 61370095, 61472124, 61202109, [18] S. Bernard, S. Adam, and L. Heutte, “Dynamic random
61472126,61672221), the National Research Foundation of forests,” Pattern Recognition Letters, vol. 33, no. 12, pp.
Qatar (NPRP, Grant Nos. 8-519-1-108), and the Natural 1580–1586, September 2012.
Science Foundation of Hunan Province of China (Grant Nos. [19] T. M. Khoshgoftaar, J. V. Hulse, and A. Napolitano, “Com-
paring boosting and bagging techniques with noisy and
2015JJ4100, 2016JJ4002).
imbalanced data,” Systems, Man and Cybernetics, IEEE
Transactions on, vol. 41, no. 3, pp. 552–568, May 2011.
R EFERENCES [20] G. Yu, N. A. Goussies, J. Yuan, and Z. Liu, “Fast action
[1] X. Wu, X. Zhu, and G.-Q. Wu, “Data mining with big detection via discriminative random forest voting and top-
data,” Knowledge and Data Engineering, IEEE Transactions k subvolume search,” Multimedia, IEEE Transactions on,
on, vol. 26, no. 1, pp. 97–107, January 2014. vol. 13, no. 3, pp. 507–517, June 2011.
[2] L. Kuang, F. Hao, and Y. L.T., “A tensor-based approach [21] G. Biau, “Analysis of a random forests model,” Journal of
for big data representation and dimensionality reduction,” Machine Learning Research, vol. 13, no. 1, pp. 1063–1095,
Emerging Topics in Computing, IEEE Transactions on, vol. 2, January 2012.
no. 3, pp. 280–291, April 2014. [22] J. D. Basilico, M. A. Munson, T. G. Kolda, K. R. Dixon, and
[3] A. Andrzejak, F. Langner, and S. Zabala, “Interpretable W. P. Kegelmeyer, “Comet: A recipe for learning and using
models from distributed data via merging of decision large ensembles on massive data,” in IEEE International
trees,” in Computational Intelligence and Data Mining (CID- Conference on Data Mining, October 2011, pp. 41–50.
M), 2013 IEEE Symposium on. IEEE, 2013, pp. 1–9. [23] K. M. Svore and C. J. Burges, “Distributed stochastic aware
[4] P. K. Ray, S. R. Mohanty, N. Kishor, and J. P. S. Catalao, random forests efficient data mining for big data,” in Big
“Optimal feature and decision tree-based classification Data (BigData Congress), 2013 IEEE International Congress
of power quality disturbances in distributed generation on. Cambridge University Press, 2013, pp. 425–426.
systems,” Sustainable Energy, IEEE Transactions on, vol. 5, [24] B. Panda, J. S. Herbach, S. Basu, and R. J. Bayardo, “Planet:
no. 1, pp. 200–208, January 2014. Massively parallel learning of tree ensembles with mapre-
[5] Apache, “Hadoop,” Website, June 2016, duce,” Proceedings of the Vldb Endowment, vol. 2, no. 2, pp.
http://hadoop.apache.org. 1426–1437, August 2009.
[6] S. del Rio, V. Lopez, J. M. Benitez, and F. Herrera, “On the [25] S. Tyree, K. Q. Weinberger, and K. Agrawal, “Parallel
use of mapreduce for imbalanced big data using random boosted regression trees for web search ranking,” in In-
forest,” Information Sciences, vol. 285, pp. 112–137, Novem- ternational Conference on World Wide Web, March 2011, pp.
ber 2014. 387–396.
[7] K. Singh, S. C. Guntuku, A. Thakur, and C. Hota, “Big [26] D. Warneke and O. Kao, “Exploiting dynamic resource
data analytics framework for peer-to-peer botnet detection allocation for efficient parallel data processing in the
using random forests,” Information Sciences, vol. 278, pp. cloud,” Parallel and Distributed Systems, IEEE Transactions
488–497, September 2014. on, vol. 22, no. 6, pp. 985–997, June 2011.
[8] Apache, “Spark,” Website, June 2016, http://spark- [27] L. Mashayekhy, M. M. Nejad, D. Grosu, Q. Zhang, and
project.org. W. Shi, “Energy-aware scheduling of mapreduce jobs for
[9] L. Breiman, “Random forests,” Machine Learning, vol. 45, big data applications,” Parallel and Distributed Systems,
no. 1, pp. 5–32, October 2001. IEEE Transactions on, vol. 26, no. 3, pp. 1–10, March 2015.
[10] G. Wu and P. H. Huang, “A vectorization-optimization- [28] L. D. Briceno, H. J. Siegel, A. A. Maciejewski, M. Oltikar,
method-based type-2 fuzzy neural network for noisy data and J. Brateman, “Heuristics for robust resource allocation
classification,” Fuzzy Systems, IEEE Transactions on, vol. 21, of satellite weather data processing on a heterogeneous
no. 1, pp. 1–15, February 2013. parallel system,” Parallel and Distributed Systems, IEEE
[11] H. Abdulsalam, D. B. Skillicorn, and P. Martin, “Classifica- Transactions on, vol. 22, no. 11, pp. 1780–1787, February
tion using streaming random forests,” Knowledge and Data 2011.
Engineering, IEEE Transactions on, vol. 23, no. 1, pp. 22–36, [29] F. Zhang, J. Cao, W. Tan, S. Khan, K. Li, and A. Zomaya,
January 2011. “Evolutionary scheduling of dynamic multitasking work-
[12] C. Lindner, P. A. Bromiley, M. C. Ionita, and T. F. Cootes, loads for big-data analytics in elastic cloud,” Emerging
“Robust and accurate shape model matching using ran- Topics in Computing, IEEE Transactions on, vol. 2, no. 3, pp.
dom forest regression-voting,” Pattern Analysis and Ma- 338–351, August 2014.
chine Intelligence, IEEE Transactions on, vol. 25, no. 3, pp. [30] K. Li, X. Tang, B. Veeravalli, and K. Li, “Scheduling
1–14, December 2014. precedence constrained stochastic tasks on heterogeneous
[13] X. Yun, G. Wu, G. Zhang, K. Li, , and S. Wang, “Fastraq: A cluster systems,” Parallel and Distributed Systems, IEEE
fast approach to range-aggregate queries in big data envi- Transactions on, vol. 64, no. 1, pp. 191–204, January 2015.
ronments,” Cloud Computing, IEEE Transactions on, vol. 3, [31] Y. Xu, K. Li, J. Hu, and K. Li, “A genetic algorithm for task
no. 2, pp. 206–218, April 2015. scheduling on heterogeneous computing systems using
[14] M. Xu, H. Chen, and P. K. Varshney, “Dimensionality multiple priority queues,” Information Sciences, vol. 270,
reduction for registration of high-dimensional data sets,” no. 6, pp. 255–287, June 2014.
Image Processing, IEEE Transactions on, vol. 22, no. 8, pp. [32] A. Spark, “Spark mllib - random forest,” Website,
3041–3049, August 2013. June 2016, http://spark.apache.org/docs/latest/mllib-
[15] Q. Tao, D. Chu, and J. Wang, “Recursive support vector ensembles.html.
machines for dimensionality reduction,” Neural Networks, [33] U. of California, “Uci machine learning repository,” Web-
IEEE Transactions on, vol. 19, no. 1, pp. 189–193, January site, June 2016, http://archive.ics.uci.edu/ml/datasets.
2008.
[16] Y. Lin, T. Liu, and C. Fuh, “Multiple kernel learning for
dimensionality reduction,” Pattern Analysis and Machine

View publication stats