2.3 AdvancedAccumulators

Author: lossyrob

project/Build.scala

@@ -46,7 +46,8 @@ object Build extends Build {
       // raise memory limits here if necessary
       javaOptions += "-Xmx4G",
 
-      mainClass in (Compile, run) := Some("com.packt.spark.section5.FeatureExtraction"),
+//      mainClass in (Compile, run) := Some("com.packt.spark.section5.FeatureExtraction"),
+      mainClass in (Compile, run) := Some("com.packt.spark.section2.AdvancedAccumulators"),
 
       fork := true,
       connectInput in run := true,

src/main/scala/com/packt/spark/section2/Accumulators.complete

@@ -6,42 +6,36 @@ import org.apache.spark._
 object Accumulators extends ExampleApp {
   def run() =
     withSparkContext { implicit sc =>
-      val infoAcc = sc.accumulable(DatasetInfo())
-
-      val violationEntries = 
-        sampleDataset
-          .flatMap { line =>
-            val parsed = Violation.fromRow(line)
-            infoAcc += parsed
-            parsed
-           }
-
-      // OR
-
-      val violationEntries = 
-        sampleDataset
-          .map(Violation.fromRow _)
-          .map(infoAcc += _)
-          .flatten
-
-      // More computations would happen here...
-      violationEntries.foreach { x =>  }
-
-      val DatasetInfo(
-        dateRange,
-        validCount,
-        invalidCount,
-        bigTicketItems,
-        totalFines
-      ) = infoAcc.value
-
-      println(s"Valid count: $validCount")
-      println(s"Invalid count: $invalidCount")
-      println(s"Date range: ${dateRange.start} to ${dateRange.end}")
-      println("Big ticket items:")
-      for( (desc, fine) <- bigTicketItems) {
-        println(s" $fine  $desc")
-      }
-
+      val validAcc = sc.accumulator(0)
+      val invalidAcc = sc.accumulator(0)
+      val totalFineAcc = sc.accumulator(0.0)
+ 
+      val violations =
+        fullDataset
+          .mapPartitions { partition =>
+            val parse = Violation.rowParser
+            partition.flatMap { line =>
+              parse(line) match {
+                case v @ Some(violation) if violation.ticket.fine > 5.0 =>
+                  validAcc += 1
+                  totalFineAcc += violation.ticket.fine
+                  v
+                case _ =>
+                  invalidAcc += 1
+                  None
+              }
+            }
+          }
+
+      violations.foreach { x => }
+
+      val validCount = validAcc.value
+      val invalidCount = invalidAcc.value
+      val totalFines = totalFineAcc.value
+
+      println(s"Valid count is $validCount")
+      println(s"Invalid count is $invalidCount")
+      println(f"Total fines: $$${totalFines}%,1.2f")
+      waitForUser()
     }
 }

src/main/scala/com/packt/spark/section2/Accumulators.scala

@@ -1,52 +1,41 @@
 package com.packt.spark.section2
 
 import com.packt.spark._
-
 import org.apache.spark._
-import com.github.nscala_time.time.Imports._
 
 object Accumulators extends ExampleApp {
-  implicit object MaxFineAccumulatorParam extends AccumulatorParam[Ticket] {
-    def zero(ticket: Ticket) = Ticket(0.0, "None")
-    def addInPlace(ticket1: Ticket, ticket2: Ticket): Ticket =
-      if(ticket1.fine > ticket2.fine) ticket1
-      else ticket2
-  }
   def run() =
     withSparkContext { implicit sc =>
       val validAcc = sc.accumulator(0)
       val invalidAcc = sc.accumulator(0)
-      val sumAcc = sc.accumulator(0.0)
-      val maxFineAcc = sc.accumulator(Ticket(0.0, "None"))
-      val dateRangeAcc = sc.accumulator(DataDateRange.empty)
-
-      val violationEntries = 
-        sampleDataset
-          .flatMap { line =>
-            Violation.fromRow(line) match {
-              case e @ Some(entry) =>
-                validAcc += 1
-                sumAcc += entry.ticket.fine
-                maxFineAcc += entry.ticket
-                dateRangeAcc += entry.issueTime
-                e
-              case None =>
-                invalidAcc += 1
-                None
+      val totalFineAcc = sc.accumulator(0.0)
+ 
+      val violations =
+        fullDataset
+          .mapPartitions { partition =>
+            val parse = Violation.rowParser
+            partition.flatMap { line =>
+              parse(line) match {
+                case v @ Some(violation) if violation.ticket.fine > 5.0 =>
+                  validAcc += 1
+                  totalFineAcc += violation.ticket.fine
+                  v
+                case _ =>
+                  invalidAcc += 1
+                  None
+              }
             }
           }
 
-      violationEntries.foreach { x =>  }
+      violations.foreach { x => }
 
       val validCount = validAcc.value
       val invalidCount = invalidAcc.value
-      val totalFines = sumAcc.value
-      val maxFineTicket = maxFineAcc.value
+      val totalFines = totalFineAcc.value
 
-      println(s"Valid count: ${validAcc.value}")
-      println(s"Invalid count: ${invalidAcc.value}")
-      println(f"Total fines: $$${sumAcc.value.toLong}%,d")
-      println(s"Max fine ticket: $maxFineTicket")
-      println(s"Date range: ${dateRangeAcc.value.start} to ${dateRangeAcc.value.end}")
+      println(s"Valid count is $validCount")
+      println(s"Invalid count is $invalidCount")
+      println(f"Total fines: $$${totalFines}%,1.2f")
+      waitForUser()
     }
 }

src/main/scala/com/packt/spark/section2/AdvancedAccumulators.complete

@@ -0,0 +1,57 @@
+package com.packt.spark.section2
+
+import com.packt.spark._
+
+import org.apache.spark._
+import com.github.nscala_time.time.Imports._
+
+object AdvancedAccumulators extends ExampleApp {
+  implicit object MaxFineAccumulatorParam extends AccumulatorParam[Ticket] {
+    def zero(ticket: Ticket) = Ticket(0.0, "None")
+    def addInPlace(ticket1: Ticket, ticket2: Ticket): Ticket =
+      if(ticket1.fine > ticket2.fine) ticket1
+      else ticket2
+  }
+  def run() =
+    withSparkContext { implicit sc =>
+      val invalidAcc = sc.accumulator(0)
+      val infoAcc = sc.accumulable(DatasetInfo())
+
+      val violationEntries = 
+        fullDataset
+          .mapPartitions { partition =>
+            val parse = Violation.rowParser
+            partition.flatMap { line =>
+              parse(line) match {
+                case v @ Some(violation) if violation.ticket.fine > 5.0 =>
+                  infoAcc += violation
+                  v
+                case _ =>
+                  invalidAcc += 1
+                  None
+              }
+            }
+          }
+
+      violationEntries.foreach { x =>  }
+
+      val DatasetInfo(
+        dateRange,
+        validCount,
+        bigTicketItems,
+        totalFines
+      ) = infoAcc.value
+
+      val invalidCount = invalidAcc.value
+
+      println(s"Valid count is $validCount")
+      println(s"Invalid count is $invalidCount")
+      println(f"Total fines: $$${totalFines}%,1.2f")
+      println(s"Date range: ${dateRange.start} to ${dateRange.end}")
+      println("Big ticket items:")
+      for( (desc, fine) <- bigTicketItems) {
+        println(s" $fine  $desc")
+      }
+      waitForUser()
+    }
+}

src/main/scala/com/packt/spark/section2/AdvancedAccumulators.scala

@@ -0,0 +1,51 @@
+package com.packt.spark.section2
+
+import com.packt.spark._
+
+import org.apache.spark._
+import com.github.nscala_time.time.Imports._
+
+object AdvancedAccumulators extends ExampleApp {
+  def run() =
+    withSparkContext { implicit sc =>
+      val invalidAcc = sc.accumulator(0)
+      val infoAcc = sc.accumulable(DatasetInfo())
+
+      val violationEntries = 
+        fullDataset
+          .mapPartitions { partition =>
+            val parse = Violation.rowParser
+            partition.flatMap { line =>
+              parse(line) match {
+                case v @ Some(violation) if violation.ticket.fine > 5.0 =>
+                  infoAcc += violation
+                  v
+                case _ =>
+                  invalidAcc += 1
+                  None
+              }
+            }
+          }
+
+      violationEntries.foreach { x =>  }
+
+      val DatasetInfo(
+        dateRange,
+        validCount,
+        bigTicketItems,
+        totalFines
+      ) = infoAcc.value
+
+      val invalidCount = invalidAcc.value
+
+      println(s"Valid count is $validCount")
+      println(s"Invalid count is $invalidCount")
+      println(f"Total fines: $$${totalFines}%,1.2f")
+      println(s"Date range: ${dateRange.start} to ${dateRange.end}")
+      println("Big ticket items:")
+      for( (desc, fine) <- bigTicketItems) {
+        println(s" $fine  $desc")
+      }
+      waitForUser()
+    }
+}

src/main/scala/com/packt/spark/section2/AdvancedAccumulators.start

@@ -0,0 +1,41 @@
+package com.packt.spark.section2
+
+import com.packt.spark._
+import org.apache.spark._
+
+object Accumulators extends ExampleApp {
+  def run() =
+    withSparkContext { implicit sc =>
+      val validAcc = sc.accumulator(0)
+      val invalidAcc = sc.accumulator(0)
+      val totalFineAcc = sc.accumulator(0.0)
+ 
+      val violations =
+        fullDataset
+          .mapPartitions { partition =>
+            val parse = Violation.rowParser
+            partition.flatMap { line =>
+              parse(line) match {
+                case v @ Some(violation) if violation.ticket.fine > 5.0 =>
+                  validAcc += 1
+                  totalFineAcc += violation.ticket.fine
+                  v
+                case _ =>
+                  invalidAcc += 1
+                  None
+              }
+            }
+          }
+
+      violations.foreach { x => }
+
+      val validCount = validAcc.value
+      val invalidCount = invalidAcc.value
+      val totalFines = totalFineAcc.value
+
+      println(s"Valid count is $validCount")
+      println(s"Invalid count is $invalidCount")
+      println(f"Total fines: $$${totalFines}%,1.2f")
+      waitForUser()
+    }
+}

src/main/scala/com/packt/spark/section2/AdvancedAccumulators.step1

@@ -0,0 +1,52 @@
+package com.packt.spark.section2
+
+import com.packt.spark._
+import org.apache.spark._
+
+object AdvancedAccumulators extends ExampleApp {
+  implicit object MaxFineAccumulatorParam extends AccumulatorParam[Ticket] {
+    def zero(ticket: Ticket) = Ticket(0.0, "None")
+    def addInPlace(ticket1: Ticket, ticket2: Ticket): Ticket =
+      if(ticket1.fine > ticket2.fine) ticket1
+      else ticket2
+  }
+
+  def run() =
+    withSparkContext { implicit sc =>
+      val validAcc = sc.accumulator(0)
+      val invalidAcc = sc.accumulator(0)
+      val totalFineAcc = sc.accumulator(0.0)
+      val maxFineAcc = sc.accumulator(Ticket(0.0, "None"))
+ 
+      val violations =
+        fullDataset
+          .mapPartitions { partition =>
+            val parse = Violation.rowParser
+            partition.flatMap { line =>
+              parse(line) match {
+                case v @ Some(violation) if violation.ticket.fine > 5.0 =>
+                  validAcc += 1
+                  totalFineAcc += violation.ticket.fine
+                  maxFineAcc += violation.ticket
+                  v
+                case _ =>
+                  invalidAcc += 1
+                  None
+              }
+            }
+          }
+
+      violations.foreach { x => }
+
+      val validCount = validAcc.value
+      val invalidCount = invalidAcc.value
+      val totalFines = totalFineAcc.value
+      val maxFineTicket = maxFineAcc.value
+
+      println(s"Valid count is $validCount")
+      println(s"Invalid count is $invalidCount")
+      println(f"Total fines: $$${totalFines}%,1.2f")
+      println(s"Max fine ticket: $maxFineTicket")
+      waitForUser()
+    }
+}