SDS-2.2, Scalable Data Science

Archived YouTube video of this live unedited lab-lecture:

Activity Recognition from Accelerometer using Random Forest

This work is a simpler databricksification of Amira Lakhal's more complex framework for activity recognition:

• https://github.com/MiraLak/activityRecognitionV2.

See Section below on Download and Load Data first.

val data = sc.textFile("dbfs:///datasets/sds/ActivityRecognition/dataTraining.csv")

data: org.apache.spark.rdd.RDD[String] = dbfs:///datasets/sds/ActivityRecognition/dataTraining.csv MapPartitionsRDD[45250] at textFile at <console>:34

data.take(5).foreach(println)

"user_id","activity","timeStampAsLong","x","y","z"
"user_001","Jumping",1446047227606,"4.33079","-12.72175","-3.18118"
"user_001","Jumping",1446047227671,"0.575403","-0.727487","2.95007"
"user_001","Jumping",1446047227735,"-1.60885","3.52607","-0.1922"
"user_001","Jumping",1446047227799,"0.690364","-0.037722","1.72382"

val dataDF = sqlContext.read    
    .format("com.databricks.spark.csv") // use spark.csv package
    .option("header", "true") // Use first line of all files as header
    .option("inferSchema", "true") // Automatically infer data types
    .option("delimiter", ",") // Specify the delimiter as ','
    .load("dbfs:///datasets/sds/ActivityRecognition/dataTraining.csv")

dataDF: org.apache.spark.sql.DataFrame = [user_id: string, activity: string ... 4 more fields]

dataDF.printSchema()

root
 |-- user_id: string (nullable = true)
 |-- activity: string (nullable = true)
 |-- timeStampAsLong: long (nullable = true)
 |-- x: double (nullable = true)
 |-- y: double (nullable = true)
 |-- z: double (nullable = true)

display(dataDF)

Truncated to 30 rows

dataDF.count()

res2: Long = 13679

dataDF.select($"user_id").distinct().show()

+--------+
| user_id|
+--------+
|user_002|
|user_006|
|user_005|
|user_001|
|user_007|
|user_003|
|user_004|
+--------+

dataDF.select($"activity").distinct().show()

+--------+
|activity|
+--------+
| Sitting|
| Walking|
| Jumping|
|Standing|
| Jogging|
+--------+

display(dataDF.sample(false,0.1))

Truncated to 30 rows

Feature Selection on Running Windows

A Markov Process Assumption

This is sensible since the subjects are not instantaneously changing between the activities of interest: sitting, walking, jogging, etc. Thus it makes sense to try and use the most recent accelerometer readings (from the immediate past) to predict the current activity.

See the following for a crash introduction to windows:

• http://xinhstechblog.blogspot.co.nz/2016/04/spark-window-functions-for-dataframes.html

 // Import the window functions.
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

// Create a window specification
val windowSize = 10
val wSpec1 = Window.partitionBy("user_id","activity").orderBy("timeStampAsLong").rowsBetween(-windowSize, 0)

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._
windowSize: Int = 10
wSpec1: org.apache.spark.sql.expressions.WindowSpec = org.apache.spark.sql.expressions.WindowSpec@69e228a1

 // Calculate the moving window statistics from data
val dataFeatDF = dataDF
      .withColumn( "meanX", mean($"x").over(wSpec1)  )
      .withColumn( "meanY", mean($"y").over(wSpec1)  )
      .withColumn( "meanZ", mean($"z").over(wSpec1)  ) 
//resultant = 1/n * ∑ √(x² + y² + z²)
      .withColumn( "SqX", pow($"x",2.0) )
      .withColumn( "SqY", pow($"y",2.0) )
      .withColumn( "SqZ", pow($"z",2.0) )
      .withColumn( "resultant", pow( $"SqX"+$"SqY"+$"SqZ",0.50 ) )
      .withColumn( "meanResultant", mean("resultant").over(wSpec1) )
// (1 / n ) * ∑ |b - mean_b|, for b in {x,y,z} 
      .withColumn( "absDevFromMeanX", abs($"x" - $"meanX") )
      .withColumn( "absDevFromMeanY", abs($"y" - $"meanY") )
      .withColumn( "absDevFromMeanZ", abs($"z" - $"meanZ") )
      .withColumn( "meanAbsDevFromMeanX", mean("absDevFromMeanX").over(wSpec1) )
      .withColumn( "meanAbsDevFromMeanY", mean("absDevFromMeanY").over(wSpec1) )
      .withColumn( "meanAbsDevFromMeanZ", mean("absDevFromMeanZ").over(wSpec1) )
//standard deviation  = √ variance = √ 1/n * ∑ (x - u)² with u = mean x
      .withColumn( "sqrDevFromMeanX", pow($"absDevFromMeanX",2.0) )
      .withColumn( "sqrDevFromMeanY", pow($"absDevFromMeanY",2.0) )
      .withColumn( "sqrDevFromMeanZ", pow($"absDevFromMeanZ",2.0) )
      .withColumn( "varianceX", mean("sqrDevFromMeanX").over(wSpec1) )
      .withColumn( "varianceY", mean("sqrDevFromMeanY").over(wSpec1) )
      .withColumn( "varianceZ", mean("sqrDevFromMeanZ").over(wSpec1) )
      .withColumn( "stddevX", pow($"varianceX",0.50) )
      .withColumn( "stddevY", pow($"varianceY",0.50) )
      .withColumn( "stddevZ", pow($"varianceZ",0.50) )

dataFeatDF: org.apache.spark.sql.DataFrame = [user_id: string, activity: string ... 27 more fields]

display(dataFeatDF.sample(false,0.1))

Truncated to 30 rows

Truncated to 12 cols

val splits = dataFeatDF.randomSplit(Array(0.7, 0.3))
val (trainingData, testData) = (splits(0), splits(1))

splits: Array[org.apache.spark.sql.Dataset[org.apache.spark.sql.Row]] = Array([user_id: string, activity: string ... 27 more fields], [user_id: string, activity: string ... 27 more fields])
trainingData: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [user_id: string, activity: string ... 27 more fields]
testData: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [user_id: string, activity: string ... 27 more fields]

See

• http://spark.apache.org/docs/latest/ml-classification-regression.html#random-forest-classifier
• http://blog.citizennet.com/blog/2012/11/10/random-forests-ensembles-and-performance-metrics

import org.apache.spark.ml.feature.{StringIndexer,VectorAssembler}
import org.apache.spark.ml.Pipeline

import org.apache.spark.ml.classification.RandomForestClassifier

val transformers = Array(
              new StringIndexer().setInputCol("activity").setOutputCol("label"),
              new VectorAssembler()
                      .setInputCols(Array("meanX", "meanY", "meanZ", "stddevX", "stddevY","stddevZ"))
                      .setOutputCol("features")
)

// Train a RandomForest model.
val rf = new RandomForestClassifier() 
              .setLabelCol("label")
              .setFeaturesCol("features")
              .setNumTrees(10)
              .setFeatureSubsetStrategy("auto")
              .setImpurity("gini")
              .setMaxDepth(20)
              .setMaxBins(32)
              .setSeed(12345)

val model = new Pipeline().setStages(transformers :+ rf).fit(trainingData)

import org.apache.spark.ml.feature.{StringIndexer, VectorAssembler}
import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.classification.RandomForestClassifier
transformers: Array[org.apache.spark.ml.PipelineStage with org.apache.spark.ml.param.shared.HasOutputCol with org.apache.spark.ml.util.DefaultParamsWritable{def copy(extra: org.apache.spark.ml.param.ParamMap): org.apache.spark.ml.PipelineStage with org.apache.spark.ml.param.shared.HasOutputCol with org.apache.spark.ml.util.DefaultParamsWritable{def copy(extra: org.apache.spark.ml.param.ParamMap): org.apache.spark.ml.PipelineStage with org.apache.spark.ml.param.shared.HasOutputCol with org.apache.spark.ml.util.DefaultParamsWritable}}] = Array(strIdx_865d012b7f68, vecAssembler_7ddfcff33c0c)
rf: org.apache.spark.ml.classification.RandomForestClassifier = rfc_531a2c5f8138
model: org.apache.spark.ml.PipelineModel = pipeline_286ffbefb82f

val accuracy: Double = 1.0 * model.transform(testData)
                                  .select("activity","label","prediction")
                                  .filter($"label"===$"prediction").count() / testData.count()

accuracy: Double = 0.9795042897998093

We get 98% correct predictions and here are the mis-predicted ones.

display(model.transform(testData).select("activity","label","prediction").filter(not($"label"===$"prediction")))

Truncated to 30 rows

Download and Load Data

The following anonymized dataset is used with kind permission of Amira Lakhal.

wget http://lamastex.org/datasets/public/ActivityRecognition/dataTraining.csv

--2017-01-09 23:39:33--  http://lamastex.org/datasets/public/ActivityRecognition/dataTraining.csv
Resolving lamastex.org (lamastex.org)... 166.62.28.100
Connecting to lamastex.org (lamastex.org)|166.62.28.100|:80... connected.
HTTP request sent, awaiting response... 200 OK
Length: 931349 (910K) [text/csv]
Saving to: ‘dataTraining.csv’

     0K .......... .......... .......... .......... ..........  5%  131K 7s
    50K .......... .......... .......... .......... .......... 10%  262K 5s
   100K .......... .......... .......... .......... .......... 16%  263K 4s
   150K .......... .......... .......... .......... .......... 21%  263K 3s
   200K .......... .......... .......... .......... .......... 27% 57.5M 3s
   250K .......... .......... .......... .......... .......... 32%  264K 2s
   300K .......... .......... .......... .......... .......... 38% 65.2M 2s
   350K .......... .......... .......... .......... .......... 43% 68.4M 1s
   400K .......... .......... .......... .......... .......... 49%  264K 1s
   450K .......... .......... .......... .......... .......... 54% 57.0M 1s
   500K .......... .......... .......... .......... .......... 60% 54.2M 1s
   550K .......... .......... .......... .......... .......... 65%  265K 1s
   600K .......... .......... .......... .......... .......... 71% 55.4M 1s
   650K .......... .......... .......... .......... .......... 76% 67.7M 0s
   700K .......... .......... .......... .......... .......... 82% 49.9M 0s
   750K .......... .......... .......... .......... .......... 87%  636K 0s
   800K .......... .......... .......... .......... .......... 93%  455K 0s
   850K .......... .......... .......... .......... .......... 98% 67.5M 0s
   900K .........                                             100% 57.1M=1.7s

2017-01-09 23:39:35 (530 KB/s) - ‘dataTraining.csv’ saved [931349/931349]

pwd && ls

/databricks/driver
dataTraining.csv
derby.log
eventlogs
logs

//dbutils.fs.mkdirs("dbfs:///datasets/sds/ActivityRecognition")
dbutils.fs.mv("file:///databricks/driver/dataTraining.csv","dbfs:///datasets/sds/ActivityRecognition/")

res13: Boolean = true

display(dbutils.fs.ls("dbfs:///datasets/sds/ActivityRecognition"))