Neural networks were introduced in Chapter 2, Apache Spark MLlib. This chapter builds upon this understanding by introducing deep learning, which uses deep neural networks. These are neural networks that are feature-rich, and contain extra hidden layers, so that their ability to extract data features is increased. These networks are generally feed-forward networks, where the feature characteristics are inputs to the input layer neurons. These neurons then fire and spread the activation through the hidden layer neurons to an output layer, which should present the feature label values. Errors in the output are then propagated back through the network (at least in back propagation), adjusting the neuron connection weight matrices so that classification errors are reduced during training.
The previous example image, described in the H2O booklet at https://leanpub.com/deeplearning/read ,shows a deep learning network with four input neurons to the left, two hidden layers in the middle, and two output neurons. The arrows show both the connections between neurons and the direction that activation takes through the network.
These networks are feature-rich because they provide the following options:
- Multiple training algorithms
- Automated network configuration
- The ability to configure many options
- Structure
Hidden layer structure
- Training
Learning rate, annealing, and momentum
- Structure
So, after giving this brief introduction to deep learning, it is now time to look at some of the sample Scala-based code. H2O provides a great deal of functionality; the classes that are needed to build and run the network have been developed for you. You just need to do the following:
- Prepare the data and parameters
- Create and train the model
- Validate the model with a second data set
- Score the validation data set output
When scoring your model, you must hope for a high value in percentage terms. Your model must be able to accurately predict and classify your data.
This section examines the Scala-based H2O Sparkling Water deep learning example using the previous Canadian income data source. First, the Spark (Context, Conf, mllib, and RDD), and H2O (h2o, deeplearning, and water) classes are imported:
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import hex.deeplearning.{DeepLearningModel, DeepLearning}
import hex.deeplearning.DeepLearningModel.DeepLearningParameters
import org.apache.spark.h2o._
import org.apache.spark.mllib
import org.apache.spark.mllib.feature.{IDFModel, IDF, HashingTF}
import org.apache.spark.rdd.RDD
import water.KeyNext an application class called h2o_spark_dl2 is defined, the master URL is created, and then a configuration object is created, based on this URL, and the application name. The Spark context is then created using the configuration object:
object h2o_spark_dl2 extends App
{
val sparkMaster = "spark://hc2nn.semtech-solutions.co.nz:7077"
val appName = "Spark h2o ex1"
val conf = new SparkConf()
conf.setMaster(sparkMaster)
conf.setAppName(appName)
val sparkCxt = new SparkContext(conf)An H2O context is created from the Spark context, and also an SQL context:
import org.apache.spark.h2o._ implicit val h2oContext = new org.apache.spark.h2o.H2OContext(sparkCxt).start() import h2oContext._ import org.apache.spark.sql._ implicit val sqlContext = new SQLContext(sparkCxt)
The H2O Flow user interface is started with the openFlow command:
import sqlContext._ openFlow
The training and testing of the data files are now defined (on HDFS) using the server URL, path, and the file names:
val server = "hdfs://hc2nn.semtech-solutions.co.nz:8020" val path = "/data/spark/h2o/" val train_csv = server + path + "adult.train.data" // 32,562 rows val test_csv = server + path + "adult.test.data" // 16,283 rows
The CSV based training and testing data is loaded using the Spark context's textFile method:
val rawTrainData = sparkCxt.textFile(train_csv) val rawTestData = sparkCxt.textFile(test_csv)
Now, the schema is defined in terms of a string of attributes. Then, a schema variable is created by splitting the string using a series of StructField, based on each column. The data types are left as String, and the true value allows for the Null values in the data:
val schemaString = "age workclass fnlwgt education “ +
“educationalnum maritalstatus " + "occupation relationship race
gender “ + “capitalgain capitalloss " + hoursperweek nativecountry income"
val schema = StructType( schemaString.split(" ")
.map(fieldName => StructField(fieldName, StringType, true)))The raw CSV line training and testing data is now split by commas into columns. The data is filtered on empty lines to ensure that the last column (income) is not empty. The actual data rows are created from the fifteen (0-14) trimmed elements in the raw CSV data. Both, the training and the test data sets are processed:
val trainRDD = rawTrainData
.filter(!_.isEmpty)
.map(_.split(","))
.filter( rawRow => ! rawRow(14).trim.isEmpty )
.map(rawRow => Row(
rawRow(0).toString.trim, rawRow(1).toString.trim,
rawRow(2).toString.trim, rawRow(3).toString.trim,
rawRow(4).toString.trim, rawRow(5).toString.trim,
rawRow(6).toString.trim, rawRow(7).toString.trim,
rawRow(8).toString.trim, rawRow(9).toString.trim,
rawRow(10).toString.trim, rawRow(11).toString.trim,
rawRow(12).toString.trim, rawRow(13).toString.trim,
rawRow(14).toString.trim
)
)
val testRDD = rawTestData
.filter(!_.isEmpty)
.map(_.split(","))
.filter( rawRow => ! rawRow(14).trim.isEmpty )
.map(rawRow => Row(
rawRow(0).toString.trim, rawRow(1).toString.trim,
rawRow(2).toString.trim, rawRow(3).toString.trim,
rawRow(4).toString.trim, rawRow(5).toString.trim,
rawRow(6).toString.trim, rawRow(7).toString.trim,
rawRow(8).toString.trim, rawRow(9).toString.trim,
rawRow(10).toString.trim, rawRow(11).toString.trim,
rawRow(12).toString.trim, rawRow(13).toString.trim,
rawRow(14).toString.trim
)
)Spark Schema RDD variables are now created for the training and test data sets by applying the schema variable, created previously for the data using the Spark context's applySchema method:
val trainSchemaRDD = sqlContext.applySchema(trainRDD, schema) val testSchemaRDD = sqlContext.applySchema(testRDD, schema)
Temporary tables are created for the training and testing data:
trainSchemaRDD.registerTempTable("trainingTable")
testSchemaRDD.registerTempTable("testingTable")Now, SQL is run against these temporary tables, both to filter the number of columns, and to potentially limit the data. I could have added a WHERE or LIMIT clause. This is a useful approach that enables me to manipulate both the column and row-based data:
val schemaRddTrain = sqlContext.sql(
"""SELECT
|age,workclass,education,maritalstatus,
|occupation,relationship,race,
|gender,hoursperweek,nativecountry,income
|FROM trainingTable """.stripMargin)
val schemaRddTest = sqlContext.sql(
"""SELECT
|age,workclass,education,maritalstatus,
|occupation,relationship,race,
|gender,hoursperweek,nativecountry,income
|FROM testingTable """.stripMargin)The H2O data frames are now created from the data. The final column in each data set (income) is enumerated, because this is the column that will form the deep learning label for the data. Also, enumerating this column forces the deep learning model to carry out classification rather than regression:
val trainFrame:DataFrame = schemaRddTrain
trainFrame.replace( trainFrame.find("income"), trainFrame.vec("income").toEnum)
trainFrame.update(null)
val testFrame:DataFrame = schemaRddTest
testFrame.replace( testFrame.find("income"), testFrame.vec("income").toEnum)
testFrame.update(null)The enumerated results data income column is now saved so that the values in this column can be used to score the tested model prediction values:
val testResArray = schemaRddTest.collect()
val sizeResults = testResArray.length
var resArray = new Array[Double](sizeResults)
for ( i <- 0 to ( resArray.length - 1)) {
resArray(i) = testFrame.vec("income").at(i)
}The deep learning model parameters are now set up in terms of the number of epochs, or iterations—the data sets for training and validation and the label column income, which will be used to classify the data. Also, we chose to use variable importance to determine which data columns are most important in the data. The deep learning model is then created:
val dlParams = new DeepLearningParameters() dlParams._epochs = 100 dlParams._train = trainFrame dlParams._valid = testFrame dlParams._response_column = 'income dlParams._variable_importances = true val dl = new DeepLearning(dlParams) val dlModel = dl.trainModel.get
The model is then scored against the test data set for predictions, and these income predictions are compared to the previously stored enumerated test data income values. Finally, an accuracy percentage is output from the test data:
val testH2oPredict = dlModel.score(schemaRddTest )('predict)
val testPredictions = toRDD[DoubleHolder](testH2oPredict)
.collect.map(_.result.getOrElse(Double.NaN))
var resAccuracy = 0
for ( i <- 0 to ( resArray.length - 1)) {
if ( resArray(i) == testPredictions(i) )
resAccuracy = resAccuracy + 1
}
println()
println( ">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>" )
println( ">>>>>> Model Test Accuracy = "
+ 100*resAccuracy / resArray.length + " % " )
println( ">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>" )
println()In the last step, the application is stopped, the H2O functionality is terminated via a shutdown call, and then the Spark context is stopped:
water.H2O.shutdown() sparkCxt.stop() println( " >>>>> Script Finished <<<<< " ) } // end application
Based upon a training data set of 32,000, and a test data set of 16,000 income records, this deep learning model is quite accurate. It reaches an accuracy level of 83 percent, which is impressive for a few lines of code, small data sets, and just 100 epochs, as the run output shows:
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>> Model Test Accuracy = 83 % >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
In the next section, I will examine some of the coding needed to process the MNIST data, even though that example could not be completed due to an H2O limitation at the time of coding.
Since the MNIST image data record is so big, it presents problems while creating a Spark SQL schema, and processing a data record. The records in this data are in CSV format, and are formed from a 28 x 28 digit image. Each line is then terminated by a label value for the image. I have created my schema by defining a function to create the schema string to represent the record, and then calling it:
def getSchema(): String = {
var schema = ""
val limit = 28*28
for (i <- 1 to limit){
schema += "P" + i.toString + " "
}
schema += "Label"
schema // return value
}
val schemaString = getSchema()
val schema = StructType( schemaString.split(" ")
.map(fieldName => StructField(fieldName, IntegerType, false)))The same general approach to deep learning can be taken to data processing as the previous example, apart from the actual processing of the raw CSV data. There are too many columns to process individually, and they all need to be converted into integers to represent their data type. This can be done in one of two ways. In the first example, var args can be used to process all the elements in the row:
val trainRDD = rawTrainData.map( rawRow => Row( rawRow.split(",").map(_.toInt): _* ))The second example uses the fromSeq method to process the row elements:
val trainRDD = rawTrainData.map(rawRow => Row.fromSeq(rawRow.split(",") .map(_.toInt)))In the next section, the H2O Flow user interface will be examined to see how it can be used to both monitor H2O and process the data.