Residual vs Fitted for investment

While I was checking graphs i can draw from a regression model I realized
one graph is extremely useful for investment.
Think you have(actually we have) thousand of used car prices. And you want to buy a car
with optimum price for investment.(In hope you will sell later)

Best Car to buy Price = Max(Expected Price by regression - Real price )

Think our regression line expects a car price to be 30K but actual advertisement price is 20K.
There are 2 possibilities.
1)Car is damaged.
2)Car owner needs urgent money and selling his car with a very low price.

If car price is 40K. I can not find a logical explanation for this. Some people
are trying to sell their used cars with more price than an-unused one. Probably
they spend for some amenities which they think so valuable.

import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.ml.linalg.Vectors

val dataset = spark.createDataFrame(
  Seq(
  (20000,2011,30000.0),
  (120000,2014,20000.0),
  (60000,2015,25000.0) ,
  (20000,2011,32000.0),
  (120000,2014,21000.0),
  (60000,2015,45000.0)   
  
  )
).toDF("km", "year", "label")

val assembler = new VectorAssembler()
  .setInputCols(Array("km", "year"))
  .setOutputCol("features")

val output = assembler.transform(dataset)
output.select("features", "label").show(false)

import org.apache.spark.ml.regression.LinearRegression

val lr = new LinearRegression()
  .setMaxIter(10)
  .setRegParam(0.3)
  .setElasticNetParam(0.8)


val lrModel = lr.fit(output)

display(lrModel, output, "fittedVsResiduals")

println(s"Coefficients: ${lrModel.coefficients} Intercept: ${lrModel.intercept}")

​



val trainingSummary = lrModel.summary

println(s"numIterations: ${trainingSummary.totalIterations}")

println(s"objectiveHistory: [${trainingSummary.objectiveHistory.mkString(",")}]")

trainingSummary.residuals.show()

println(s"RMSE: ${trainingSummary.rootMeanSquaredError}")

println(s"r2: ${trainingSummary.r2}")

Coefficients: [-0.19283786035452538,2928.112739446878] Intercept: -5853577.79139608
numIterations: 8
objectiveHistory: [0.5,0.446369757257132,0.352850077757605,0.272318835721877,0.26365142412164966,0.23726105027025182,0.23725993458647637,0.23725993457463043]
+-------------------+
|          residuals|
+-------------------+
|-1000.1704245014116|
|-500.72260739002377|
| -9999.106968107633|
|  999.8295754985884|
| 499.27739260997623|
| 10000.893031892367|
+-------------------+

Residual = Observed value - Predicted value

We must find the ones with most negative residual.(Much cheaper than expected).
Databricks graph has bad resolution for few points so i wrote R version also.

library(lattice) 
mydata2 = data.frame(
  year = c(2011.0,2012.0,2014.0,2015.0),
  km10000 = c(6.0,7.0,10.0,3.0),
  price1000 = c(200.0,250.0,300.0,400.0)
)



res2.mod1 = lm(price1000 ~  km10000 + year , data = mydata2)
summary(res.mod1)
fitted(res2.mod1)
xyplot(resid(res2.mod1) ~ fitted(res2.mod1),
       xlab = "Fitted Values",
       ylab = "Residuals",
       main = "Car price based on year and km ",
       par.settings = simpleTheme(col=c("blue","red"),
                                  pch=c(10,3,11), cex=3, lwd=2),
       
       panel = function(x, y, ...)
       {
         panel.grid(h = -1, v = -1)
         panel.abline(h = 0)
         panel.xyplot(x, y, ...)
       }
)       

> fitted(res2.mod1)
       1        2        3        4 
206.1722 240.9232 302.5415 400.3631 
> resid(res2.mod1)
         1          2          3          4 
-6.1721992  9.0767635 -2.5414938 -0.3630705 
> 

Spark Dataframe Broadcast Join

Think you have a person table for your company with 10000 records. And you have 100 departments.

Problem) Find average age of people in Finance department

If we make a join,data needs to be shuffled for joining. But if we know one table is so small(like department) we can mark it as
broadcast so that our big table does not get shuffled.

Also when finding average age for department , all related data will be filtered on local nodes.

import org.apache.spark.sql.functions.broadcast

case class Person(id:Long,name: String, depid: Long)
val personDF = Seq(Person(1,"Andy", 1) ,Person(2,"John", 1),Person(3,"Jack", 2),Person(4,"Max", 2)).toDF()

case class Department(depid:Long,name: String)
val departmentDF = Seq(Department(1,"Finance") ,Department(2,"It")).toDF()

val partitionedPerson = personDF.repartition($"name").explain(true)
val combinedDF = personDF.join(departmentDF, Seq("depid"))
val combinedDF2 = personDF.join(broadcast(departmentDF), Seq("depid"))
combinedDF2.take(10)

We can also define our join with sql syntax.

personDF.createOrReplaceTempView("person")
departmentDF.createOrReplaceTempView("department")

sql("SELECT * FROM person r JOIN department s ON r.depid = s.depid").show()

id|name|depid|depid| name
1|Andy| 1| 1|Finance
2|John| 1| 1|Finance
3|Jack| 2| 2| It
4| Max| 2| 2| It

Spark HashPartitioner for RDD

RDD is a parallel distribution of data. According to our target calculation
we do not have to care how data is distributed. But if our data will benefit from data locality,
we must ensure data locality.
Think we have some car data (Honda,Toyota,Ford) and we will count,average,sum... of cars by Model.
If we have 3 partitions which these models are randomly distributed, data must be shuffled before reducing.

If we partition by car model and we are sure that all Honda data is at same partition, we will
calculate everything in one node and no need to shuffle.

When I was testing partitioning I taught partitioning will be based on data cardinality.
Since I have 3 distinct keys, their hash will be distributed on 3 slots.
But it was wrong. In fact hashcode of object is applied a mod operator on partition number.
So you could get very unpredictable results if you hash over Strings. I advise to convert Strings
to hashcode and see int value for exactly knowing final partition distribution.

val cars = Array("Honda", "Toyota", "Ford")

val carnamePrice = sc.parallelize(for {
    x <- cars
    y <- Array(100,200,300)
} yield (x, y), 8)
val rddEachCar = carnamePrice.partitionBy(new HashPartitioner(3))
val mapped =   rddEachCar.mapPartitionsWithIndex{
                    (index, iterator) => {
                       println("Called in Partition -> " + index)
                       val myList = iterator.toList
                       
                       myList.map(x => x + " -> " + index).iterator
                    }
                 }
mapped.take(10)

Array[String] = Array((Toyota,100) -> 0, (Toyota,200) -> 0, (Toyota,300) -> 0, (Honda,100) -> 1, (Honda,200) -> 1, (Honda,300) -> 1, (Ford,100) -> 2, (Ford,200) -> 2, (Ford,300) -> 2)

println ( "Honda".hashCode() % 3)
println ( "Ford".hashCode() % 3 )
println ( "Toyota".hashCode() % 3 )
1
2
0

Impute Outliers In Spark

Imputing data in R is so easy.

fun <- function(x){
    quantiles <- quantile( x, c(.05, .95 ) )
    x[ x < quantiles[1] ] <- quantiles[1]
    x[ x > quantiles[2] ] <- quantiles[2]
    x
}
fun( yourdata )

I tried to write my functions for Spark. Below you will find methods for replacing values
Q1 is 1st quantile and
Q3 is 3rd quantile
left outliers < Q1 – 1.5×IQR
right outliers > Q3 + 1.5×IQR
For a dataset I obtain below values for these results.

org.apache.spark.util.StatCounter = (count: 13, mean: 131.538462, stdev: 65.441242, max: 320.000000, min: 1.000000) 
rddMin: Double = 1.0 
rddMax: Double = 320.0 
rddMean: Double = 131.53846153846155 
quantiles: Array[Double] = Array(117.0, 150.0) 
Q1: Double = 117.0 
Q3: Double = 150.0 
IQR: Double = 33.0 
lowerRange: Double = 67.5 
upperRange: Double = 199.5

So according to our data set and our logic we can make below changes.(Of course there are more)
left outliers -> Q1
left outliers -> mean
right outliers -> Q3
right outliers -> mean
You can also want to impute values less than Q1 to Q1.

import org.apache.spark.sql.functions._

val imputeLessToTarget: (Double,Double,Double ) => Double = (arg: Double,treshold: Double,target: Double) => { if (arg < treshold) target else arg}
val imputeLessToTargetUDF = udf(imputeLessToTarget)

val imputeMoreToTarget: (Double,Double,Double ) => Double = (arg: Double,treshold: Double,target: Double) => { if (arg > treshold) target else arg}
val imputeMoreToTargetUDF = udf(imputeMoreToTarget)


//Change values less than lowerRange of outliers to Q1 **** Capping
val df5 = df.withColumn("replaced", imputeLessToTargetUDF(col("original") ,lit(lowerRange),lit(Q1) ))
println( "df5 Change values less than lowerRange of outliers to Q1 **** Capping")
df5.show()
//Change values less than lowerRange of outliers to mean
val df6 = df.withColumn("replaced", imputeLessToTargetUDF(col("original") ,lit(lowerRange),lit(rddMean) ))
println( "df6 Change values less than lowerRange of outliers to mean")
df6.show()
//Change values less than Q1 to mean
val df7 = df.withColumn("replaced", imputeLessToTargetUDF(col("original") ,lit(Q1),lit(rddMean) ))
println( "df7 Change values less than Q1 to mean")
df7.show()
//Change values more than upperRange of outliers to Q3 **** Capping
val df8 = df.withColumn("replaced", imputeMoreToTargetUDF(col("original") ,lit(upperRange),lit(Q3) ))
println( "df8 Change values more than upperRange of outliers to Q3 **** Capping")
df8.show()
//Change values more than upperRange of outliers to mean
val df9 = df.withColumn("replaced", imputeMoreToTargetUDF(col("original") ,lit(upperRange),lit(rddMean) ))
println( "df9 Change values more than upperRange of outliers to mean")
df9.show()

At below you can see output of generating new columns. These type of functions are my toolset for dealing with new data.

df5 Change values less than lowerRange of outliers to Q1 **** Capping 
1.0| 117.0
110.0| 110.0
111.0| 111.0
112.0| 112.0
117.0| 117.0
118.0| 118.0
120.0| 120.0
122.0| 122.0
129.0| 129.0
140.0| 140.0
150.0| 150.0
160.0| 160.0
320.0| 320.0
df6 Change values less than lowerRange of outliers to mean 
1.0|131.53846153846155
110.0| 110.0
111.0| 111.0
112.0| 112.0
117.0| 117.0
118.0| 118.0
120.0| 120.0
122.0| 122.0
129.0| 129.0
140.0| 140.0
150.0| 150.0
160.0| 160.0
320.0| 320.0
df7 Change values less than Q1 to mean 
1.0|131.53846153846155
110.0|131.53846153846155
111.0|131.53846153846155
112.0|131.53846153846155
117.0| 117.0
118.0| 118.0
120.0| 120.0
122.0| 122.0
129.0| 129.0
140.0| 140.0
150.0| 150.0
160.0| 160.0
320.0| 320.0
df8 Change values more than upperRange of outliers to Q3 **** Capping
1.0| 150.0
110.0| 150.0
111.0| 150.0
112.0| 150.0
117.0| 150.0
118.0| 150.0
120.0| 150.0
122.0| 150.0
129.0| 150.0
140.0| 150.0
150.0| 150.0
160.0| 150.0
320.0| 320.0
df9 Change values more than upperRange of outliers to mean 
1.0|131.53846153846155
110.0|131.53846153846155
111.0|131.53846153846155
112.0|131.53846153846155
117.0|131.53846153846155
118.0|131.53846153846155
120.0|131.53846153846155
122.0|131.53846153846155
129.0|131.53846153846155
140.0|131.53846153846155
150.0|131.53846153846155
160.0|131.53846153846155
320.0| 320.0

SparkException, Task not serializable

I usually run my codes on databricks for a 1st test.I was facing with lots of "Task not serializable" exceptions.
I was solving them in a way. Then I googled some and tried to list every reason that can cause this exception.
Here I will write simplest one.
When writing on databricks(single class lets say) if at a point i try to


Let's define 3 classes. Exception1 is class throwing serializable exception.
The reason behind this is when there is a call to function, Spark tries to serialize the enclosing object class.
There are 2 solutions.

1) NoException2 extends java.io.Serializable
just make class Serializable

2) Make add function function with def.
def add(a:Int) = a+1 // function
instead of
val add = (a: Int) => a + 1 // method

Functions in scala are serializable.

class Exception1 {

    val rdd = sc.parallelize(List(1,2,3))

    def addFunc =  {
      val result = rdd.map(add)
      result.take(3)
    }

    def add(a:Int) = a+1

  }
class NoException1 {
  val rdd = sc.parallelize(List(1,2,3))

  def addFunc() =  {
    val result = rdd.map(add)
    result.take(3)
  }

  val add = (a: Int) => a + 1
}
class NoException2 extends java.io.Serializable {
  val rdd = sc.parallelize(List(1,2,3))

  def doIT() =  {
    val result = rdd.map(add)
    result.take(3)
  }

  def add(a: Int) = a + 1
}

Calling functions.

(new Exception1()).addFunc
(new NoException1()).addFunc
(new NoException2()).addFunc

Results

org.apache.spark.SparkException: Task not serializable
res9: Array[Int] = Array(2, 3, 4)
res9: Array[Int] = Array(2, 3, 4)

Write Elastic from Kafka

Reading and writing from Kafka is so time consuming because you write your code,
wait data to come. If you see exception break the program and write again.

Our task was
1)Read a Json formatted data from kafka
2)Filter messages if they include some keywords.
3)Write to elastic search.

**Add related kafka streaming jars to your Zeppelin or
application.

import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._
import org.apache.spark.rdd.RDD
import scalaj.http._
import org.apache.spark.sql.SQLContext
import com.stratio.datasource.util.Config._
import org.apache.spark.sql.functions._
import org.elasticsearch.spark.sql._
import scalaj.http._
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.functions._

val ssc = new StreamingContext(sc, Seconds(30))
val topicMap:Map[String,Int] =Map("YourTopic"->1)
val lines = KafkaUtils.createStream(ssc,"your.com:2181", "group", topicMap).map(_._2)

lines.foreachRDD((rdd: RDD[String], time: Time) => {
      
  val df2 = sqlContext.read.json(rdd)

 
  val df = df2.filter($"msg".contains("mykeyword1") || $"msg".contains("mykeyword1") )
 
  
  val esConfig = Map("es.nodes"->"192.168.1.151","es.port"->"9200")

         df.saveToEs("elastic/target",esConfig)
 
    })

Spark, Migrate Data from Solr to MongoDB in Batches

Most of big data projects are migration. And most of utility packages does not work as in samples because
of version differences , library conflicts.

We decided to use Stratio/Spark-MongoDB for Spark to MongoDB migration.

We had to move data from old Solr files to Mongo. If we try to move all data it was getting exceptions
which implies size problems.(Sorry forgot the exceptions.) Then we need a method for writing the items in batch.
We tried different sizes and found 50000 was working.

Solr was also returning how many records are there actually even if we do not fetch.
So we defined a function(of course we could do at loop) to check if we consumed all of them in batches.

We got some errors on usage of library stratio. I found standard codes were not compatible with our environment and
found magic configuration by checking source code. I omitted lots of code below but it gives a perfect idea.
And in fact it is runnable if you write your data frames correctly.
Our environment was Spark 1.6.2 .

import scalaj.http._
import com.mongodb.casbah.{WriteConcern => MongodbWriteConcern}
import com.stratio.datasource._
import com.stratio.datasource.mongodb._
import com.stratio.datasource.mongodb.schema._
import com.stratio.datasource.mongodb.writer._
import com.stratio.datasource.mongodb.config._
import com.stratio.datasource.mongodb.config.MongodbConfig._
import org.apache.spark.sql.SQLContext
import com.stratio.datasource.util.Config._
import org.apache.spark.sql.functions._

val wbuilder = MongodbConfigBuilder(Map(Host -> List("192.168.2.150:27018"), Database -> "ManagementDev", Collection ->"arsiv", SamplingRatio -> 1.0, WriteConcern -> "normal", ReadPreference -> "Secondary"))
val writeConfig = wbuilder.build()

var condition=true
var start=0
var pass=0

def checkPaging(total:Long,start:Int):Boolean={
    if((start+50000)<(total)) {
        return true
    }
    else {
        return false
    }
}

do{
    val response1 = Http("http://192.168.2.155:8983/solr/collection1/select?q=*%3A*&sort=tw_id+asc&start="+start+"&rows=50000&fl=AuthorEmail&wt=json")
    .timeout(connTimeoutMs = 1000000, readTimeoutMs = 500000).header("key", "val").method("get")
    .execute().body
    val rdd1 = sc.parallelize(Seq(response1))
 
    val df1 = sqlContext.read.json(rdd1)
    
    val df2=df1.select($"response.docs")
    
    //Lots of omitted codes
    val dcount=df1.select("response.numFound").take(1)(0)(0).asInstanceOf[Long]
    //Lots of omitted codes

    dfy.write.format("parquet").mode(org.apache.spark.sql.SaveMode.Append).partitionBy("saveCriteria").save("hdfs://your.com/user/Archive") 
    dfx.saveToMongodb(writeConfig)
    
    condition=checkPaging(dcount,start)
    start=start+50000
    pass=pass+1
    println("!!PASS "+pass )
} 
while(condition)

Gradient Descent for Linear Regression

I was searching for something(I do not remember) and I saw below post.
GradientDescentExample

This was a perfect post to test some parameters on gradient descent.
I opened my databricks notebook and began to play with it. I took the functions from this page but changed a bit
because of type changes in my code.

Question : Check picture below. We have points as below, is there a formula
that identifies this spread.

I used y1 = 5 * x1 + 10 + noise formula to generate this data.
So our target values are 5 and 10. ( Or a little different because of noise)

noise = np.random.normal(-3, 6, 49)

x1 = np.linspace(0, 50, 49)
y1 = 5 * x1 + 10 + noise
points = zip(x1,y1)

fig3, ax3 = plt.subplots()
#ax3.plot(x1, y1, 'k--')

ax3.plot(x1, y1, 'ro')



display(fig3)

So lets say you made an initial guess.
y = 5 * x + 3



Now lets say this is your initial guess. We must calculate how good is y = 5 * x + 3
from sum( (guess - actual)^2 ) / len ( standard formula)

What is next step? Make a better guess. How do determine you will be making a better guess.
There must be a function which will determine how your error decreases. Gradient descent function will
help you choose better values for slope and intercept.

There are some parameters you are giving to function. Learning rate and iteration count.
Learning rate is hard to understand. I checked various learning rates to see the effect.

for learning_rate = 0.00001
As you see in the graph error function is diminishing at each run. But by time
improvement is getting smaller.

for learning_rate = 0.001
It seems it is learning faster. But be careful this is a simple example and our distribution is simple with only one minimum.
If we had a complex function who has lots of convex ,concave shapes then our high/low learning rates could skip global minimum
or stuck in local minimum. Check pictures in net for these effects. There are lots of nice pictures.

You can play with parameters below and you will obtain very different results according to your parameters.

num_tests = 10
mycoef = 1
iter_count = 500

learning_rate = 0.001
initial_b = 0 # initial y-intercept guess
initial_m = 0 # initial slope guess


figError, axError = plt.subplots()

fig2, ax2 = plt.subplots()
plt.figure(1)

ax2.plot(x1, y1, 'ro')

errorList = [];


for i in range(num_tests):
  [b, m] = gradient_descent_runner(points, initial_b, initial_m, learning_rate, (i+mycoef)* iter_count)  
  
  
  x = np.linspace(0, 50, 49)
  y = x *m + b
  ax2.plot(x, y, 'k--')
  ax2.text(max(x),max(y),i)

  error = compute_error_for_line_given_points(b, m, points)
  errorList.append( error );
  
  
  axError.plot( i ,error , 'bo')
  print "After {0} iterations b = {1}, m = {2}, error = {3}".format( (i+mycoef)* iter_count, b, m, error)

Below is a result of parameters
num_tests = 10
mycoef = 1
iter_count = 500

You can play as much as you want and see the of error, slope and intercept.
Below graph seems bad because ,function performs so good from beginning and lines overlap.

Part I took from article

from numpy import *
import numpy as np
from StringIO import StringIO
import matplotlib.pyplot as plt
import numpy as np

# y = mx + b
# m is slope, b is y-intercept
def compute_error_for_line_given_points(b, m, points):
    totalError = 0
    for i in range(0, len(points)):
        x = points[i][ 0]
        y = points[i][ 1]
        totalError += (y - (m * x + b)) ** 2
    return totalError / float(len(points))

def step_gradient(b_current, m_current, points, learningRate):
    b_gradient = 0
    m_gradient = 0
    N = float(len(points))
    for i in range(0, len(points)):
        x = points[i, 0]
        y = points[i, 1]
        b_gradient += -(2/N) * (y - ((m_current * x) + b_current))
        m_gradient += -(2/N) * x * (y - ((m_current * x) + b_current))
    new_b = b_current - (learningRate * b_gradient)
    new_m = m_current - (learningRate * m_gradient)
    return [new_b, new_m]

def gradient_descent_runner(points, starting_b, starting_m, learning_rate, num_iterations):
    b = starting_b
    m = starting_m
    for i in range(num_iterations):
        b, m = step_gradient(b, m, array(points), learning_rate)
    return [b, m]

Spark and R K-Means classification

***
Spark samples are for big files which contains thousands of lines.
Also you do not know data and can not play with it.
I put here simplest data set for spark mllib so that one can play and understand what metrics
are effected from which parameters.
It is not for seniors but perfect for beginners of who need to calibrate parameters with simple sets.
***
Below code is from sample Spark documentation. I changed Rdd so that one can play and understand
how data is distributed.




Here you can play with values and observe distribution of clusters.
Always print cluster centers. It will give you a clue for large datasets.

You can easily play with dataset and number of demanded clusters to get an idea of how
K-means work.

import org.apache.spark.mllib.clustering.{KMeans, KMeansModel}
import org.apache.spark.mllib.linalg.Vectors

val parsedData = sc.parallelize(Seq(
  ( Vectors.dense(1.0, 1.0)),
  ( Vectors.dense(40.0, 40.0)),
  ( Vectors.dense(60.0, 60.0)),
  ( Vectors.dense(101.0, 101.1))
))

// Cluster the data into two classes using KMeans
val numClusters = 2
val numIterations = 20
val clusters = KMeans.train(parsedData, numClusters, numIterations)

// Evaluate clustering by computing Within Set Sum of Squared Errors
val WSSSE = clusters.computeCost(parsedData)
println("Within Set Sum of Squared Errors = " + WSSSE )
val clusterCenters = clusters.clusterCenters.map(_.toArray)
println("The Cluster Centers are = " + clusterCenters)
parsedData.collect().map( s=> println( "cluster "+clusters.predict(s) +" "+s.toString() ) )

Result

Within Set Sum of Squared Errors = 3874.8066666666673
clusterCenters: Array[Array[Double]] = Array(Array(67.0, 67.03333333333333), Array(1.0, 1.0))
cluster 1 [1.0,1.0] cluster 0 [40.0,40.0] cluster 0 [60.0,60.0] cluster 0 [101.0,101.1]

Same Code In R

pointx = c(1,2, 50, 51) 
pointy = c(1,2,50,51) 
df = data.frame(pointx, pointy)
library(ggplot2)
ggplot(df, aes(pointx, pointy)) + geom_point()
myCluster <- kmeans(df, 3, nstart = 20)
myCluster$centers
myCluster$clus <- as.factor(myCluster$cluster)
ggplot(df, aes(pointx, pointy, color = myCluster$clus)) + geom_point()

Spark NaiveBayes and Result Interpretation

***
Spark samples are for big files which contains thousands of lines.
Also you do not know data and can not play with it.
I put here simplest data set for spark mllib so that one can play and understand what metrics
are effected from which parameters.
It is not for seniors but perfect for beginners of who need to calibrate parameters with simple sets.
***
In samples at internet people usually try to guess if a mail is spam or not.
Below code includes codes from spark samples and some other samples.
I tried to work with spark 2 but it was not success. It is working with 1.6.
Since it did not work i played a lot, took lots of fixes from net. So code is not neat.

Lets make it much more simpler. I will list some properties and try to guess if it is
Plane or Not.
My training is
"wing wheel engine" : 1 it is plane
"wheel airbag engine" : 0 it is not plane

Steps
1)Get training set
2)Tokenize it
3)Apply hashingtf

Result of hashingtf, it generates 2vectors of words.

0 wheel airbag engine ["wheel","airbag","engine"] {"type":0,"size":20,"indices":[3,14,18],"values":[1,1,1]}
1 wing wheel engine ["wing","wheel","engine"] {"type":0,"size":20,"indices":[3,7,14],"values":[1,1,1]}

4)Train model
Result of training

Array[org.apache.spark.mllib.regression.LabeledPoint] = Array(
(8.0,[0.0,0.0,0.0,0.0,0.0,0.0,0.0,1.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,1.0,0.0]), 
(9.0,[0.0,0.0,0.0,1.0,0.0,0.0,0.0,1.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0])
)

5)Prepare test data
(0,"wing airbag")
(1,"wing airport")
(0,"wing airport") False negative(this will be guest as plane, but it is zeppelin!!)
hashingtf generates below vectors for test data

[8,wing airbag,WrappedArray(wing, airbag),(20,[7,18],[1.0,1.0])], 
[9,wing airport,WrappedArray(wing, airport),(20,[3,7],[1.0,1.0])])

7 was wing and 3 was wheel in model vectors.

6)Apply prediction testpredictionAndLabel
(0.0,0.0) I guessed as not plane, Not plane
(1.0,1.0) I guessed as plane , Plane
(1.0,0.0) I guessed as plane , Plane

7)Dump metrics, output is as below

Confusion matrix: 
1.0 1.0 
0.0 1.0 
Precision(0.0) = 1.0 
Precision(1.0) = 0.5 
Recall(0.0) = 0.5 
Recall(1.0) = 1.0 
FPR(0.0) = 0.0 
FPR(1.0) = 0.5 
F1-Score(0.0) = 0.6666666666666666 
F1-Score(1.0) = 0.6666666666666666 
Weighted precision: 0.8333333333333333 
Weighted recall: 0.6666666666666666 
Weighted F1 score: 0.6666666666666666 
Weighted false positive rate: 0.16666666666666666 
labels: Array[Double] = Array(0.0, 1.0)

Precision (0.0) :1.0 we guest 1 zero(false) , that was correct so ratio 1 / 1 = 1
Precision (1.0) :0.5 we guest 2 one(true) , 1 was correct 1 not so ratio 1 / 2 = 0.5

Recall(0.0) :0.5 we guest 1 zero(false) , there was infact 2 zeros 1 / 2 = 0.5
Recall(1.0) :1.0 we guest 1 one(true) , there was correct 1 not so ratio 1 / 1 = 1

F1-Score(0.0) = 0.6666666666666666

F1- Score = 2 x ( precision x recall ) / precison + recall.
= 2 x ( 0.5 x 1 ) / 0.5 + 1 = 2 x 0.5 / 1.5 = 0.6

From definitions :
Precision can be seen as a measure of exactness or quality, whereas recall is a measure of completeness or quantity.
High precision means that an algorithm returned substantially more relevant results than irrelevant ones, while high recall means that an algorithm returned most of the relevant results.

What does these mean.
Think in our sample we have a bigger set and we say
there are 30 planes but only 20 of them is really (among 30)
then precision is 20 / 30 = this is how well we performed on our results.
But there are items we missed.
Think in fact there were total 50 planes.
Then recall = 20 / 50 = 0.4
It is what percent of real result we returned.


High precision Low recall : we are very good at estimation but we do not cover the whole space. It means
we choose cut-off value so high.

import org.apache.spark.ml.feature.{RegexTokenizer, Tokenizer}
import org.apache.spark.ml.feature.{HashingTF, IDF}
import org.apache.spark.mllib.classification.{NaiveBayes, NaiveBayesModel}
import org.apache.spark.mllib.util.MLUtils
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.linalg.Vector
import org.apache.spark.mllib.evaluation.MulticlassMetrics



val trainData = sqlContext.createDataFrame(Seq((0,"wheel airbag engine"),(1,"wing wheel engine"))).toDF("category","text")
    val tokenizer = new Tokenizer().setInputCol("text").setOutputCol("words")
    val wordsData = tokenizer.transform(trainData)
    val hashTF = new HashingTF().setInputCol("words").setOutputCol("features").setNumFeatures(20)
    val featureData = hashTF.transform(wordsData) 
val subFeature = featureData.select("category","features");
val df_1 = subFeature.withColumnRenamed("category","category2")
val trainDataRdd2 = df_1.withColumn("category",df_1.col("category2").cast("double")).drop("category2")


trainDataRdd2.printSchema()
val testScoreAndLabel = trainDataRdd2.select("category","features").map{ case Row(l:Double,p:Vector) => LabeledPoint(l,p) }

    val model = NaiveBayes.train(testScoreAndLabel, lambda = 1.0, modelType = "multinomial")
   //same for the test data
    val testData = sqlContext.createDataFrame(Seq((0,"wing airbag"),(1,"wing airport"),(0,"wing airport"))).toDF("category","text")
    val testWordData = tokenizer.transform(testData)
    val testFeatureData = hashTF.transform(testWordData)
    val testDataRdd = testFeatureData.select("category","features").map {
    case Row(label: Int, features: Vector) =>
    LabeledPoint(label.toDouble, Vectors.dense(features.toArray))
    }
    val testpredictionAndLabel = testDataRdd.map(p => (model.predict(p.features), p.label))


val metrics = new MulticlassMetrics(testpredictionAndLabel)
/* output F1-measure for all labels (0 and 1, negative and positive) */
metrics.labels.foreach( l => println(metrics.fMeasure(l)))
testpredictionAndLabel.take(5)
// Confusion matrix
println("Confusion matrix:")
println(metrics.confusionMatrix)


// Precision by label
val labels = metrics.labels
labels.foreach { l =>
  println(s"Precision($l) = " + metrics.precision(l))
}

// Recall by label
labels.foreach { l =>
  println(s"Recall($l) = " + metrics.recall(l))
}

// False positive rate by label
labels.foreach { l =>
  println(s"FPR($l) = " + metrics.falsePositiveRate(l))
}

// F-measure by label
labels.foreach { l =>
  println(s"F1-Score($l) = " + metrics.fMeasure(l))
}

// Weighted stats
println(s"Weighted precision: ${metrics.weightedPrecision}")
println(s"Weighted recall: ${metrics.weightedRecall}")
println(s"Weighted F1 score: ${metrics.weightedFMeasure}")
println(s"Weighted false positive rate: ${metrics.weightedFalsePositiveRate}")
  

Spark BinaryClassificationMetrics

After finishing a LogisticRegression we can check if result is good with BinaryClassificationMetrics.
It simply takes 2 parameters.
One is score associated with your predicition.(rawPrediction column after a Logistic Regression) for example.
And other is what you guesses.
For ROC you must have a big area near 1.

What does this mean.
Suppose you are measuring if you use heater according to weather.
(of course this is obvious, we are now doing obvious case)

Say at 10 F : do not use
20 F : do not use
...
50 F : use
..
100 F : use

You see for low scores ,u do not use, but for high ones you use.
True and false it perfectly separated so I expect a perfect ROC.

ROC is a graph showing what we gain as data for calculations we did with Logistic Regression.
for example we can have 4 data for one point. If you check below it means we only use heater once on this period.
So value 10 give 3 0 and 1 1 value. This makes learning of value 10 less efficient.
Intervals must give as much as information as possible.
Purified intervals will only output 1 value for so that information gain. is so high.

( 10.0, 0.0),
( 10.0, 0.0),
( 10.0, 0.0),
( 10.0, 1.0),

val metricData= sc.parallelize(
  
   Seq( 
     ( 10.0,  0.0),
     ( 20.0,  0.0),
     ( 30.0,  0.0),
     ( 40.0,  0.0),
     ( 50.0,  0.0),
     ( 60.0,  1.0),
     ( 70.0,  1.0),
     ( 80.0,  1.0),
     ( 90.0,  1.0),     
     ( 100.0,  1.0)
    
     )
);

val metrics = new BinaryClassificationMetrics(metricData) 
println("area under the precision-recall curve: " + metrics.areaUnderPR)
println("area under the receiver operating characteristic (ROC) curve : " + metrics.areaUnderROC)
metrics.roc().collect()

Above case was so good so metrics are below.

area under the precision-recall curve: 1.0 
area under the receiver operating characteristic (ROC) curve : 0.9999999999999999 
Array[(Double, Double)] = Array((0.0,0.0), (0.0,0.2), (0.0,0.4), (0.0,0.6), (0.0,0.8), (0.0,1.0), (0.2,1.0), (0.4,1.0), (0.6,1.0), (0.8,1.0), (1.0,1.0), (1.0,1.0))

Lets preapre a bad data where distribution is useless.
Think you are measuring your ice-tea consumption according to weather.
As in above you do need have a pattern. You do not drink at 10F but you drink 20 ...
So this is near random distribution. And random distribution gives 0.5 area under curve.
It is 45 degree line. A line like that means on every probability(score) of event
I have equal info from True or False case.

val metricData= sc.parallelize(
  
   Seq( 
     ( 10.0,  0.0),
     ( 20.0,  1.0),
     ( 30.0,  0.0),
     ( 40.0,  1.0),
     ( 50.0,  0.0),
     ( 60.0,  1.0),
     ( 70.0,  0.0),
     ( 80.0,  1.0),
     ( 90.0,  0.0),     
     ( 100.0,  1.0)
    
     )
);

val metrics = new BinaryClassificationMetrics(metricData) 
println("area under the precision-recall curve: " + metrics.areaUnderPR)
println("area under the receiver operating characteristic (ROC) curve : " + metrics.areaUnderROC)
metrics.roc().collect()

Above case was so bad so metrics are below.

area under the precision-recall curve: 0.6393650793650794 
area under the receiver operating characteristic (ROC) curve : 0.6000000000000001 metrics: 
Array[(Double, Double)] = Array((0.0,0.0), (0.0,0.2), (0.2,0.2), (0.2,0.4), (0.4,0.4), (0.4,0.6), (0.6,0.6), (0.6,0.8), (0.8,0.8), (0.8,1.0), (1.0,1.0), (1.0,1.0))

Spark Apply Descriptive Statistics on DataFrame

When you first get your data you have to play with it.
You want to learn what kind of data you have.
Below is a simple code piece to begin investigating general properties of your data.

Suppose you have a data like 48,49,50,51,52. This is well distributed homogenous data.

import org.apache.commons.math3.stat.descriptive._

val df = Seq(48,49.0, 50.0, 51.0,52.0).toDF("nums")

val mean = df.select("nums").rdd.map(row => row(0).asInstanceOf[Double]).collect()

val arrMean = new DescriptiveStatistics()
genericArrayOps(mean).foreach(v => arrMean.addValue(v))

val meanQ1 = arrMean.getPercentile(25)
val meanQ3 = arrMean.getPercentile(75)
val meanIQR = meanQ3 - meanQ1

Perfect distribution
47,48,49,50,51

n: 5 
min: 48.0 max: 52.0 mean: 50.0 
std dev: 1.5811388300841898 
median: 50.0 
skewness: 0.0 
kurtosis: -1.200000000000002 
meanQ1: Double = 48.5 
meanQ3: Double = 51.5 
meanIQR: Double = 3.0

Lets form a line shaped distribution
val df = Seq(50,50, 50, 50,50.0).toDF("nums")

n: 5 
min: 50.0 
max: 50.0 
mean: 50.0 
std dev: 0.0 
median: 50.0 
skewness: NaN 
kurtosis: NaN 
meanQ1: Double = 50.0 
meanQ3: Double = 50.0 
meanIQR: Double = 0.0

Lets add 40 to make left skew(negative skew.
** Skewness is asymmetry of distribution about mean.



Left tail (skew ) distribution

val df = Seq(40,48,49, 50, 51,52.0).toDF("nums")

n: 6 
min: 40.0 
max: 52.0 
mean: 48.333333333333336 
std dev: 4.320493798938574 
median: 49.5 
skewness: -1.8805720776629977 
kurtosis: 3.9187500000000064 
meanQ1: Double = 46.0 
meanQ3: Double = 51.25 
meanIQR: Double = 5.25

Rigth tail (skew ) distribution
If we just add 60 to original series we get a right tail distribution.
Skewness is same with different sign.
val df = Seq(48,49, 50, 51,52.0,60).toDF("nums")

n: 6 
min: 48.0 
max: 60.0 
mean: 51.666666666666664 
std dev: 4.320493798938574 
median: 50.5 
skewness: 1.8805720776629975 
kurtosis: 3.9187500000000064

meanQ1: Double = 48.75 
meanQ3: Double = 54.0 
meanIQR: Double = 5.25

meanIQR is a data without boundaries. So it gives lots of idea if you know your domain.
For example you have a car price data. You know that car must be around 50.000$.
When you check meanIQR you will see datas near to your expectation. Others will
have have meaningless high( irreal expectation of seller) or low( this time meaningful because
car could be damaged.) meanIQR is a nice measure.

Skewness can give a rough idea about tendency of data. (Data having a tail to left if minus.)

kurtosis is a measure of shape. The sharper the top the higher the kurtosis. Check picture from internet please.