Feature/scala code/ch04 deepak (#14)

* ch04-averagebykeyusereducebykey * ch04-averagebykeyusegroupbykey * ch04-averagebykeyusecombinebykey * ch04-averagebykeyusecombinebykey * ch04-DFMedianExact

Author: deepakmca05

code/chap04/scala/build.gradle

@@ -15,21 +15,12 @@ repositories {
 
 dependencies {
  implementation "org.scala-lang:scala-library:$scalaVersion"
- compile "org.scala-lang:scala-compiler:$scalaVersion"
  implementation "org.apache.spark:spark-core_$scalaClassifier:$sparkVersion"
  implementation "org.apache.spark:spark-sql_$scalaClassifier:$sparkVersion"
- implementation 'org.scalanlp:breeze_2.13:2.0.1-RC2'
+ implementation "org.scalanlp:breeze_$scalaClassifier:2.0.1-RC2"
 
 }
 
 application {
  mainClass = project.hasProperty("mainClass") ? project.getProperty("mainClass") : "NULL"
-}
-
-// To start the scala repl use the command ./gradlew repl --console plain --no-daemon
-task repl(type:JavaExec) {
- main = "scala.tools.nsc.MainGenericRunner"
- classpath = sourceSets.main.runtimeClasspath
- standardInput System.in
- args '-usejavacp'
 }

code/chap04/scala/src/main/scala/org/data/algorithms/spark/ch04/AverageByKeyUseCombineByKey.scala

@@ -0,0 +1,59 @@
+package org.data.algorithms.spark.ch04
+
+import org.apache.spark.sql.SparkSession
+
+object AverageByKeyUseCombineByKey {
+ def createSparkSession: SparkSession = {
+ SparkSession.builder()
+ .master("local[*]")
+ .appName("AverageByKeyUseReduceByKey")
+ .getOrCreate()
+ }
+
+ def main(args: Array[String]): Unit = {
+ //Create a new spark session
+ val spark = createSparkSession
+ val input = List(
+ ("k1", 1), ("k1", 2), ("k1", 3), ("k1", 4), ("k1", 5),
+ ("k2", 6), ("k2", 7), ("k2", 8),
+ ("k3", 10), ("k3", 12)
+ )
+ // build RDD<key, value>
+ val rdd = spark.sparkContext.parallelize(input)
+ /*
+ Combined data structure (C) is a Tuple2(sum, count)
+ 3 functions needs to be defined:
+ v --> C
+ C, v --> C
+ C, C --> C
+ */
+ val sumCount = rdd.combineByKey(
+ v=> (v,1),
+ (C:(Int,Int),v:Int) => (C._1+v , C._2 + 1) ,
+ (C1:(Int,Int),C2:(Int,Int)) => (C1._1+C2._1,C1._2+C2._2)
+ )
+ //show sum count
+ println(s"sum_count = ${sumCount.collect().mkString("["," ,","]")}")
+ /*
+ [
+ (k3, (22, 2)),
+ (k2, (21, 3)),
+ (k1, (15, 5))
+ ]
+ */
+
+ //find averages
+ val avg = sumCount.mapValues(values => (values._1/values._2).toFloat)
+ println(s"avg = ${avg.collect().mkString("["," ,","]")}")
+ /*
+ avg.collect()
+ [
+ (k3, 11.0),
+ (k2, 7.0),
+ (k1, 3.0)
+ ]
+ */
+ //Done!
+ spark.stop()
+ }
+}

code/chap04/scala/src/main/scala/org/data/algorithms/spark/ch04/AverageByKeyUseGroupByKey.scala

@@ -0,0 +1,50 @@
+package org.data.algorithms.spark.ch04
+
+import org.apache.spark.sql.SparkSession
+
+object AverageByKeyUseGroupByKey {
+ def createSparkSession: SparkSession = {
+ SparkSession.builder()
+ .master("local[*]")
+ .appName("AverageByKeyUseReduceByKey")
+ .getOrCreate()
+ }
+
+ def main(args: Array[String]): Unit = {
+ // create an instance of SparkSession
+ val spark = createSparkSession
+ val input = List(
+ ("k1", 1), ("k1", 2), ("k1", 3), ("k1", 4), ("k1", 5),
+ ("k2", 6), ("k2", 7), ("k2", 8),
+ ("k3", 10), ("k3", 12)
+ )
+ // build RDD<key, value>
+ val rdd = spark.sparkContext.parallelize(input)
+ //group (key, value) pairs by key
+ val groupByKey = rdd.groupByKey()
+ //show grouped by key
+ println(s"grouped_by_key = ${(groupByKey.mapValues(values => values.toList).collect()).mkString("["," , ","]")}")
+ /*
+ [
+ ('k3', List(10, 12)),
+ ('k2', List(6, 7, 8)),
+ ('k1', List(1, 2, 3, 4, 5))
+ ]
+ */
+
+ //find averages
+ val avg = groupByKey.mapValues(values => values.sum.toFloat / values.size.toFloat )
+ println(s"avg = ${avg.collect().mkString("["," ,","]")}")
+ /*
+ avg.collect()
+ [
+ (k3, 11.0),
+ (k2, 7.0),
+ (k1, 3.0)
+ ]
+ */
+ //Done!
+ spark.stop()
+ }
+
+}

code/chap04/scala/src/main/scala/org/data/algorithms/spark/ch04/AverageByKeyUseReduceByKey.scala

@@ -0,0 +1,67 @@
+package org.data.algorithms.spark.ch04
+
+import org.apache.spark.sql.SparkSession
+
+object AverageByKeyUseReduceByKey {
+
+ def createSparkSession: SparkSession = {
+ SparkSession.builder()
+ .master("local[*]")
+ .appName("AverageByKeyUseReduceByKey")
+ .getOrCreate()
+ }
+
+ def main(args: Array[String]): Unit = {
+ //Create a new spark session
+ val spark = createSparkSession
+ val input = List(
+ ("k1", 1), ("k1", 2), ("k1", 3), ("k1", 4), ("k1", 5),
+ ("k2", 6), ("k2", 7), ("k2", 8),
+ ("k3", 10), ("k3", 12)
+ )
+ // build RDD<key, value>
+ val rdd = spark.sparkContext.parallelize(input)
+ // map each (key, value) into (key, (value, 1))
+ val pairs = rdd.map( kv => (kv._1, (kv._2, 1)))
+ /*
+ pairs =
+ [
+ ("k1", (1, 1)), ("k1", (2, 1)), ("k1", (3, 1)), ("k1", (4, 1)), ("k1", (5, 1)),
+ ("k2", (6, 1)), ("k2", (7, 1)), ("k2", (8, 1)),
+ ("k3", (10, 1)), ("k3", (12, 1))
+ ]
+
+ reduce by key:
+ x = (sum1, count1)
+ y = (sum2, count2)
+ x + y --> (sum1+sum2, count1+count2)
+ */
+ val sumCount = pairs.reduceByKey((x,y)=>(x._1+y._1,x._2+y._2))
+ println(s"sum_count = ${sumCount.collect().mkString("[", ", ", "]")}")
+ /*
+ [
+ ('k3', (22, 2)),
+ ('k2', (21, 3)),
+ ('k1', (15, 5))
+ ]
+
+ find averages
+ v = (sum-of-values, count-of-values)
+ v[0] = sum-of-values
+ v[1] = count-of-values
+ */
+ val avg = sumCount.mapValues(v => (v._1/v._2).toFloat)
+ print(s"avg = ${avg.collect().mkString("[", ", ", "]")}")
+ /*
+ avg.collect()
+ [
+ ('k3', 11),
+ ('k2', 7),
+ ('k1', 3)
+ ]
+ */
+
+ // done!
+ spark.stop()
+ }
+}

code/chap04/scala/src/main/scala/org/data/algorithms/spark/ch04/DataframeMedianExact.scala

@@ -0,0 +1,65 @@
+package org.data.algorithms.spark.ch04
+
+import org.apache.spark.sql.functions.{col, collect_list, rand, udf}
+import org.apache.spark.sql.{DataFrame, SparkSession}
+
+/*
+-----------------------------------------------------
+ This program find the exact median per key.
+
+------------------------------------------------------
+ Note-1: print(), collect(), and show() are used
+ for debugging and educational purposes only.
+ 
+ ------------------------------------------------------
+ Input Parameters:
+ none
+ -------------------------------------------------------
+ 
+ @author Deepak Kumar
+ -------------------------------------------------------
+*/
+object DataframeMedianExact {
+
+ def createTestDataframe(sparkSession: SparkSession, numOfKeys: Int, numOfRows: Int): DataFrame = {
+ val key = (col("id") % numOfKeys).alias("key")
+ val value = (rand(41)+key * numOfKeys).alias("value")
+ val df = sparkSession.range(0,numOfRows,1,1).select(key,value)
+ return df
+ }
+
+ def calculateMedian(L: Seq[Float]) : Float = {
+ breeze.stats.median(L)
+ }
+
+ def main(args: Array[String]): Unit = {
+ //create an instance of spark session
+ val spark =
+ SparkSession.builder()
+ .appName("DataframeMedianExact")
+ .master("local[*]")
+ .getOrCreate()
+ /*
+ create a DataFrame with 1000,000 rows and two columns: "key" and "value"
+ number of keys will be 10 {0, 1, 2,, ..., 9}
+ */
+ val df = createTestDataframe(spark, 10, 1000000)
+ print("df.count()=", df.count())
+ df.printSchema()
+ df.show(20, truncate=false)
+ /*
+ create a UDF from a scala function:
+ FloatType() is a return type of function calculate_median(list)
+ */
+ val calculateMedianUDF = udf( L => calculateMedian(L))
+ // relative error = 1/10,000,000
+ // use approximation df2.agg(collect_list('age'))
+ val exactMedianPerKey = df.groupBy("key").agg(calculateMedianUDF(collect_list("value")).alias("median"))
+ print("exact_median_per_key.count()=", exactMedianPerKey.count())
+ exactMedianPerKey.printSchema()
+ exactMedianPerKey.show(truncate=false)
+ //Done
+ spark.stop()
+ }
+
+}