9.1.3.1 Setting the Cores

Using the doParallel library in R, we can set the number of cores of the CPU, which you want your machine to use in while running the model. There are algorithmic ways to decide (beyond the scope of this book) how many cores you should be using if a dedicated machine for such processing is available. However, if it’s your personal machine, don’t overkill the system by using many cores. Keep in mind that assigning all the cores to this process could crash your other processes due to insufficient resources. To be safer, we used the c-2 cores, where c is the number of cores available in your machine.

library(doParallel)

# Find out how many cores are available (if you don't already know)                    
c =detectCores()
c
 [1] 4
# Find out how many cores are currently being used                    
getDoParWorkers()
 [1] 1
# Create cluster with c-2 cores                    
cl <-makeCluster(c-2)

# Register cluster                    
registerDoParallel(cl)

# Find out how many cores are being used                    
getDoParWorkers()
 [1] 2

9.1.3.2 Problem Statement

The data being used here builds a model, which can predict whether a customer would default in repaying the bank loan or not (a binary classifier) using random forest. For this demonstration, we are simply looking for the time it takes to build the model when executed in serial versus parallel manners.

 Problem : Identifying Risky Bank Loans

setwd("C:\Users\Karthik\Dropbox\Book Writing - Drafts\Chapter Drafts\Chapter 9 - Scalable Machine Learning and related technology\Datasets")
credit <-read.csv("credit.csv")
str(credit)
 'data.frame':    1000 obs. of  17 variables:
  $ checking_balance    : Factor w/ 4 levels "< 0 DM","> 200 DM",..: 1 3 4 1 1 4 4 3 4 3 ...
  $ months_loan_duration: int  6 48 12 42 24 36 24 36 12 30 ...
  $ credit_history      : Factor w/ 5 levels "critical","good",..: 1 2 1 2 4 2 2 2 2 1 ...
  $ purpose             : Factor w/ 6 levels "business","car",..: 5 5 4 5 2 4 5 2 5 2 ...
  $ amount              : int  1169 5951 2096 7882 4870 9055 2835 6948 3059 5234 ...
  $ savings_balance     : Factor w/ 5 levels "< 100 DM","> 1000 DM",..: 5 1 1 1 1 5 4 1 2 1 ...
  $ employment_duration : Factor w/ 5 levels "< 1 year","> 7 years",..: 2 3 4 4 3 3 2 3 4 5 ...
  $ percent_of_income   : int  4 2 2 2 3 2 3 2 2 4 ...
  $ years_at_residence  : int  4 2 3 4 4 4 4 2 4 2 ...
  $ age                 : int  67 22 49 45 53 35 53 35 61 28 ...
  $ other_credit        : Factor w/ 3 levels "bank","none",..: 2 2 2 2 2 2 2 2 2 2 ...
  $ housing             : Factor w/ 3 levels "other","own",..: 2 2 2 1 1 1 2 3 2 2 ...
  $ existing_loans_count: int  2 1 1 1 2 1 1 1 1 2 ...
  $ job                 : Factor w/ 4 levels "management","skilled",..: 2 2 4 2 2 4 2 1 4 1 ...
  $ dependents          : int  1 1 2 2 2 2 1 1 1 1 ...
  $ phone               : Factor w/ 2 levels "no","yes": 2 1 1 1 1 2 1 2 1 1 ...
  $ default             : Factor w/ 2 levels "no","yes": 1 2 1 1 2 1 1 1 1 2 ...
# create a random sample for training and test data                    
# use set.seed to use the same random number sequence as the tutorial                    
set.seed(123)
train_sample <-sample(1000, 900)

str(train_sample)                                                                                                              
  int [1:900] 288 788 409 881 937 46 525 887 548 453 ...
# split the data frames                    
credit_train <-credit[train_sample, ]
credit_test  <-credit[-train_sample, ]

9.1.3.3 Building the model: Serial

Note the time it takes to execute the random forest model in a serial fashion on the training data created.

 Training a model on the data

library(randomForest)

#Sequential Execution                    
system.time(rf_credit_model <-randomForest(credit_train[-17],
                                            credit_train$default,
ntree =1000))
    user  system elapsed
     1.8     0.0     1.8                                                                          

9.1.3.4 Building the Model: Parallel

In the parallel version of the code, instead of directly using the random forest model with ntree = 1000 parameters (which means build 1000 decision trees), we are going to use the foreach function with %dopar%, so we can split the 1000-decision tree building process into four processes. Each part builds 250 decision trees using the randomForest function.

#Parallel Execution                  
system.time(
  rf_credit_model_parallel <-foreach(nt =rep(250,4),
.combine = combine ,
.packages ='randomForest')
                            %dopar%
randomForest(
                              credit_train[-17],
                              credit_train$default,
ntree = nt))
    user  system elapsed
    0.33    0.09    1.73

9.1.3.5 Stopping the Clusters

Stop all the clusters and resume the execution in a serial fashion.

#Shutting down cluster - when you're done, be sure to close #the parallel backend using                  
stopCluster(cl)

Observe here, approximately, that the parallel execution is 80% faster (it might differ based on your system) than the sequential one. If a single machine using multi-cores could bring such a huge improvement, imagine the time and resources you’d save when using a large computing cluster.

Notes:

• The “user time” is the CPU time charged for the execution of user instructions of the calling process.

• The “system time” is the CPU time charged for execution by the system on behalf of the calling process.

In the next section, we go a little deeper into the Hadoop ecosystem and demonstrate the first “hello world” example using Hadoop and R.

9.2.1.1 MapReduce Example: Word Count

Imagine there is a news aggregator application trying to build an automatic topic generator for all their articles in the web. The first step in the topic generator algorithm is to build a bag-of-word with their frequencies or, in other words, count the number of occurrences of each word in an article. Since there are an enormous number of articles on the web, it definitely requires huge computational power to be able to build this topic generator.

Figure 9-4 shows the MapReduce execution flow as the article is split into many key-value pairs, processed by the Map function, which generates the intermediate key-value pair of word and a value of 1. Another process called shufflemoves the output of map to the Reducer, where finally the values are added for each keyword.

Figure 9-4. Word count example using MapReduce

Notes:

• The example needs a Linux/UNIX machine to run.

• Appropriate system paths need to be defined by the administrators.

• Here is the system information in which the code was executed.

  1. platform: i686-redhat-linux-gnu

  2. arch: i686

  3. os: Linux-gnu

  4. system: i686, Linux-gnu

  5. major: 3

  6. minor: 1.2

  7. year: 2014

  8. month: 10

  9. day: 31

  10. svn rev: 66913

  11. language: R

  12. version.string: R version 3.1.2 (2014-10-31)

• The appropriate Hadoop version is required to run the code. This code runs on Hadoop version 2.2.0, build 1529768. Comparability of this code with the latest version of Hadoop is not checked.

You must set the environment variable with the location of the Hadoop bin folder and the Hadoop streaming JAR.

Sys.setenv(HADOOP_CMD="/usr/lib/hadoop-2.2.0/bin/hadoop")
Sys.setenv(HADOOP_STREAMING="/usr/lib/hadoop-2.2.0/share/hadoop/tools/lib/hadoop-streaming-2.2.0.jar")

Then you install and call the libraries rmr2 and rhdfs. Once they are successful, you initialize the HDFS to read or write data from HDFS.

library(rmr2)
library(rhdfs)

# Hadoop File Operations                    

#initialize File HDFS                    
hdfs.init()

Then you put some sample data into the HDFS using the put() function in the rhdfs library.

#Put File into HDFS                  
hdfs.put("/home/sample.txt","/hadoop_practice")
 [1] TRUE

The you define the Map and Reduce function. This code snippet defines the way the Map and Reduce function are going to scan the text file and tokenize (a term generally given to splitting a given sentence or doc by a separator like space) into key-value pairs for counting.

# Reads a bunch of lines at a time                    
#Map Phase                    
map <-function(k,lines) {
  words.list <-strsplit(lines, '\s+')
  words <-unlist(words.list)
return( keyval(words, 1) )
}

#Reduce Phase                    
reduce <-function(word, counts) {
keyval(word, sum(counts))
}

#MapReduce Function                    
wordcount <-function (input, output=NULL) {
mapreduce(input=input, output=output, input.format="text", map=map, reduce=reduce)
}

The you run the wordcount. The wordcount function we defined is now ready to be executed on Hadoop. Before calling the function, ensure that you have set the base directory where the input file exists and where you want to put the output generated by the wordcount function:

 read text files from folder input on HDFS
 save result in folder output on HDFS
 Submit job

basedir <- '/hadoop_practice'
infile  <-file.path(basedir, 'sample.txt')
outfile <-file.path(basedir, 'output')
ret <-wordcount(infile, outfile)

Fetch the results. Once the execution of the wordcount function is complete, you can fetch the results back into R and convert that into a data frame and sort the results, as shown in this code snippet.

 Fetch results from HDFS
result <-from.dfs(outfile)
results.df <-as.data.frame(result, stringsAsFactors=F)
colnames(results.df) <-c('word', 'count')
tail(results.df,100)

          word count
 1           R     1
 2          Hi     1
 3          to     1
 4         All     1
 5        with     1
 6       class     1
 7      hadoop     3
 8     Welcome     1
 9 integrating     1

head(results.df[order(results.df$count, decreasing =TRUE),])

     word count
 7 hadoop     3
 1      R     1
 2     Hi     1
 3     to     1
 4    All     1
 5   with     1                                                                          

Since the entire book is written in R, we have presented this example of word count where R integrates with Hadoop using its Hadoop streaming library which is built in the packages rhdfc and rmr2. For demonstration sake, such an integration might be fine but in a real production system, it might not be a robust solution. Other programming languages like Java, Scala, and Python have a robust production level code integrity and a tight coupling with the Hadoop framework. In the coming sections, we will introduce the basics of Hive, Pig, and Hbase, and conclude with a real-world example using Spark.

9.2.2.1 Creating Tables

The query looks very similar to the traditional SQL query; however, what happens in the background is a lot different in Hive. Upon successful execution of this query, a new file is created in the HDFS in the default database of Hive warehouse (see Figure 9-5).

Figure 9-5. The Hive create table command

Figure 9-6 shows the emp_info table in the folder structure /user/hive/warehouse/ of HDFS.

Figure 9-6. Hive table in HDFS

9.2.2.2 Describing Tables

Once the table is created, you can use the describe formatted emp_info; command to see the structure of the table matching the one we used during creation. Along with column name, it also shows the data type of the column (see Figure 9-7).

Figure 9-7. The describe table command

9.2.2.3 Generating Data and Storing it in a Local File

The table is now ready to be loaded with some data. We have shown in Figure 9-8 the generation of some dummy data and storing it in the local directory in a file named emp_info .

Figure 9-8. Generate data and store in local file

9.2.2.4 Loading the Data into the Hive Table

Once we have the data in a local file, using the command lo ad data local inpath '/home/training/emp_info' into table emp_info; we will load the data into the Hive table emp_info in HDFS.

Figure 9-9. Load the data into a Hive table

Figure 9-10 shows the data in the HDFS file that we loaded from the local file system.

Figure 9-10. Data in the HDFS file

9.2.2.5 Selecting a Query

Figure 9-11 shows two varieties of the select query. The first one is without a where clause and the second one uses where dep = 'A'. Notice how the MapReduce framework built into Hadoop comes into play in the Hive query. This is the exact reason why we associate tools like Hive with the Hadoop ecosystem. The only difference here, unlike with the Word count example, is that we don't have to explicitly define any Map or Reduce methods; instead Hive automatically does that for us.

Figure 9-11. Select query with and without a where clause

Apart from these basic commands, Hive supports data partitioning, table joins, multi-inserts, user-defined functions, and data export. These functionality are comprehensive enough for analytical databases to be migrated into Hive.

9.2.3.1 Connecting to Pig

The command pig -x local connects to a local file system. Simply using the command pig in the terminal will connect to the HDFS. For our word count example, we will stick with the local file system

Figure 9-12. Connecting to Pig using local file system

9.2.3.2 Loading the Data

The command A1 = load '/home/training/wc.txt' as (line:chararray); will scan the file and store each line and a character array. The dump A1 command will output the following:

(Hi All Welcome to Hadoop )
(Hadoop class integrating with R Hadoop)

Figure 9-13. Load data into A1

9.2.3.3 Tokenizing Each Line

Tokenize each line into a word and store it as a list. The dump A2 command will output the following:

({(Hi),(All),(Welcome),(to),(Hadoop)})
({(Hadoop),(class),(integrating),(with),(R),(Hadoop)})

Figure 9-14. Tokenize each line

9.2.3.4 Flattening the Tokens

The A3 = foreach A2 generate flatten(tokens) as words; command will further break each tokenized line into token of words. The dump A3 command will output the following:

(Hi)
(All)
(Welcome)
(to)
(Hadoop)
(Hadoop)
(class)
(integrating)
(with)
(R)
(Hadoop)

Figure 9-15. Flattening the tokens

9.2.3.5 Grouping the Words

Using the command A4 = group A3 by words; will create a key-value pair of words and the list of the word repeated as many times as it is contained in the tokenized list. The dump A4 command will output the following:

(R,{(R)})
(Hi,{(Hi)})
(to,{(to)})
(All,{(All)})
(with,{(with)})
(class,{(class)})
(Hadoop,{(Hadoop),(Hadoop),(Hadoop)})
(Welcome,{(Welcome)})
(integrating,{(integrating)})

Figure 9-16. Group words

9.2.3.6 Counting and Sorting

The following two commands will generate the key-value pair of a word and the number of its occurrence in the document and subsequently sort by count.

• A5 = foreach A4 generate group,COUNT(A3);

• A6 = order A5 by $1 desc;

The dump A6 command will output the following:

(Hadoop,3)
(R,1)
(Hi,1)
(to,1)
(All,1)
(with,1)
(class,1)
(Welcome,1)
(integrating,1)

Figure 9-17. Starting HBase

Using Pig, many such analytical workflows involving selection, filter, join, union, sorting, grouping, and transformation could be created with ease on large datasets.

9.2.4.1 Starting HBase

Start the HBase using the shell script start-hbase.sh. Run the following three commands:

1. cd /usr/lib/hbase/

2. sudo bin/start-hbase.sh

3. hbase shell

Figure 9-18. Create and put data

9.2.4.2 Creating the Table and Put Data

The following commands will create a table named employee with two columns called details and salary. And in the details column family, it will put the data under the name and gender column.

1. create 'employee','details','salary'

2. put 'employee','e1','details:name','karthik'

3. put 'employee','e1','details:gender','m'

4. put 'employee','e1','salary:sal','20000'

Figure 9-19. Scan the data

9.2.4.3 Scanning the Data

Using the command scan 'employee, you can see how the data is stored in HBase. Each row corresponds to the column values under a column family.

A comprehensive reference guide on HBase could be found at http://hbase.apache.org/book.html#arch.overview .

9.4.3.1 Running the Demo

The function demo runs all at once and outputs the entire output at one go. However, for better understanding of what the function does, we have split the output and explained each part in detail.

# Run Deep learning demo                  
demo(h2o.deeplearning)

The demo runs.

9.4.3.2 Loading the Testing Data

Load the data from the local file system directory of R, where the H2O package is installed. It might look like C:UsersKarthikDocumentsRwin-library3.2h2oextdata.

 > prostate.hex = h2o.uploadFile(path = system.file("extdata", "prostate.csv", package="h2o"), destination_frame = "prostate.hex")                                                                              

Summary output. The summary output should match the feature definition as per Figure 9-20.

 > summary(prostate.hex)
  ID               CAPSULE          AGE             RACE           
  Min.   :  1.00   Min.   :0.0000   Min.   :43.00   Min.   :0.000  
  1st Qu.: 95.75   1st Qu.:0.0000   1st Qu.:62.00   1st Qu.:1.000  
  Median :190.50   Median :0.0000   Median :67.00   Median :1.000  
  Mean   :190.50   Mean   :0.4026   Mean   :66.04   Mean   :1.087  
  3rd Qu.:285.25   3rd Qu.:1.0000   3rd Qu.:71.00   3rd Qu.:1.000  
  Max.   :380.00   Max.   :1.0000   Max.   :79.00   Max.   :2.000  
  DPROS           DCAPS           PSA               VOL            
  Min.   :1.000   Min.   :1.000   Min.   :  0.300   Min.   : 0.00  
  1st Qu.:1.000   1st Qu.:1.000   1st Qu.:  4.900   1st Qu.: 0.00  
  Median :2.000   Median :1.000   Median :  8.664   Median :14.20  
  Mean   :2.271   Mean   :1.108   Mean   : 15.409   Mean   :15.81  
  3rd Qu.:3.000   3rd Qu.:1.000   3rd Qu.: 17.063   3rd Qu.:26.40  
  Max.   :4.000   Max.   :2.000   Max.   :139.700   Max.   :97.60  
  GLEASON        
  Min.   :0.000  
  1st Qu.:6.000  
  Median :6.000  
  Mean   :6.384  
  3rd Qu.:7.000  
  Max.   :9.000  

Model building. The function h2o.deeplearning builds a deep learning model using the response variable CAPSULE and the rest of the variable as a predictor. Additional parameters are:

• Hidden, which specifies the hidden layer sizes,

• Activation, which specifies the type of activation function; the demo uses a Tanh function

• epochs, which directs the neural network with “How many times the dataset should be iterated (streamed)”

 > # Set the CAPSULE column to be a factor column then build model.
 > prostate.hex$CAPSULE = as.factor(prostate.hex$CAPSULE)

 > model = h2o.deeplearning(x = setdiff(colnames(prostate.hex), c("ID","CAPSULE")), y = "CAPSULE", training_frame = prostate.hex, activation = "Tanh", hidden = c(10, 10, 10), epochs = 10000)                                                                                      

Print the output. The output of the five layer deep neural network is printed by accessing the model summary field from the model object created in the previous step.

 > print(model@model$model_summary)
 Status of Neuron Layers: predicting CAPSULE, 2-class classification, bernoulli distribution, CrossEntropy loss, 322 weights/biases, 8.4 KB, 3,800,000 training samples, mini-batch size 1

   layer units    type dropout       l1       l2 mean_rate rate_rms
 1     1     7   Input  0.00 %                                     
 2     2    10    Tanh  0.00 % 0.000000 0.000000  0.004538 0.009754
 3     3    10    Tanh  0.00 % 0.000000 0.000000  0.007007 0.011632
 4     4    10    Tanh  0.00 % 0.000000 0.000000  0.003262 0.005256
 5     5     2 Softmax         0.000000 0.000000  0.002906 0.000392

   momentum mean_weight weight_rms mean_bias bias_rms
 1                                                   
 2 0.000000   -0.118311   1.642809 -0.152061 1.519672
 3 0.000000    0.018304   1.594797 -0.470666 0.681625
 4 0.000000   -0.063209   1.924838 -0.545838 0.903191
 5 0.000000    0.495293   4.894484  0.012870 2.835105

Make the prediction. Since the dataset was small, we haven’t split the data into training or testing datasets, but rather show the predictions on the same dataset used in training. However, in cases where sufficient data is available, you are encouraged to run the prediction on the testing dataset to better understand the efficacy of the model.

 > # Make predictions with the trained model with training data.
 > predictions = predict(object = model, newdata = prostate.hex)

 > # Export predictions from H2O Cluster as R dataframe.
 > predictions.R = as.data.frame(predictions)

 > head(predictions.R)
   predict           p0           p1
 1       0 9.984036e-01 1.596373e-03
 2       0 9.999973e-01 2.683004e-06
 3       0 9.731078e-01 2.689217e-02
 4       0 9.496504e-01 5.034956e-02
 5       0 9.996701e-01 3.298716e-04
 6       1 4.167409e-07 9.999996e-01

 > tail(predictions.R)
     predict          p0           p1
 375       0 0.999999999 7.078566e-10
 376       0 0.986077940 1.392206e-02
 377       0 0.998982044 1.017956e-03
 378       1 0.008513801 9.914862e-01
 379       0 1.000000000 5.989944e-11
 380       0 1.000000000 2.681686e-14
                                                                                       

Model evaluation. The accuracy of the model is 99.5%, which is exceptionally good. The other measures in the output were discussed in detail throughout Chapter 6. For example, MSE, Mean Square Error (MSE), Gini index, and so on.

 > # Check performance of classification model.
 > performance = h2o.performance(model = model)

 > print(performance)
 H2OBinomialMetrics: deeplearning
 ** Reported on training data. **
 ** Metrics reported on full training frame **

 MSE:  0.01764182
 RMSE:  0.1328225
 LogLoss:  0.0741766
 Mean Per-Class Error:  0.01861449
 AUC:  0.9958826
 Gini:  0.9917653

 Confusion Matrix for F1-optimal threshold:
          0   1    Error    Rate
 0      223   4 0.017621  =4/227
 1        3 150 0.019608  =3/153
 Totals 226 154 0.018421  =7/380

 Maximum Metrics: Maximum metrics at their respective thresholds
                         metric threshold    value idx
 1                       max f1  0.347034 0.977199 114
 2                       max f2  0.347034 0.979112 114
 3                 max f0point5  0.730649 0.983718 106
 4                 max accuracy  0.551164 0.981579 110
 5                max precision  1.000000 1.000000   0
 6                   max recall  0.007983 1.000000 152
 7              max specificity  1.000000 1.000000   0
 8             max absolute_mcc  0.347034 0.961761 114
 9   max min_per_class_accuracy  0.347034 0.980392 114
 10 max mean_per_class_accuracy  0.347034 0.981386 114

More demos in the H2O package. Running the following command will list all the available demos in H2O, which you can run once and then observe how the model building process is being followed for the specific ML algorithm.

demo(package = “h2o)                    
Demos in package ‘h2o’:                                                                                      

h2o.anomaly                  H2O anomaly using prostate cancer data
h2o.deeplearning             H2O deeplearning using prostate cancer data
h2o.gbm                      H2O generalized boosting machines using prostate cancer data
h2o.glm                      H2O GLM using prostate cancer data
h2o.glrm                     H2O GLRM using walking gait data
h2o.kmeans                   H2O K-means using prostate cancer data
h2o.naiveBayes               H2O naive Bayes using iris and Congressional voting data
h2o.prcomp                   H2O PCA using Australia coast data
h2o.randomForest             H2O random forest classification using iris data