100 evaluate par prog approach for data import

build.gradle

@@ -1,5 +1,17 @@
+buildscript {
+    repositories {
+        maven {
+            url "https://plugins.gradle.org/m2/"
+        }
+    }
+    dependencies {
+        classpath "me.champeau.gradle:jmh-gradle-plugin:0.3.1"
+    }
+}
+
 subprojects {
     apply plugin: 'scala'
+    apply plugin: 'me.champeau.gradle.jmh'
 
     repositories {
         jcenter()
@@ -32,4 +44,18 @@ subprojects {
         testCompile 'com.typesafe.akka:akka-stream-testkit_2.11:2.4.12'
         testRuntime 'org.scala-lang.modules:scala-xml_2.11:1.0.5'
     }
+
+    sourceSets {
+        jmh {
+            scala.srcDir 'src/jmh/scala'
+        }
+    }
+
+    jmh {
+        include = '.*'
+        jmhVersion = '1.18'
+        jvmArgs = "-DbuildDir=$buildDir"
+        duplicateClassesStrategy = 'warn'
+        zip64 = true
+    }
 }

course-5/capstone/src/jmh/scala/observatory/bearmug/ServiceLocatorBenchmark.scala

@@ -0,0 +1,62 @@
+package observatory.bearmug
+
+import java.io.PrintWriter
+import java.time.LocalDate
+import java.util.concurrent.TimeUnit
+
+import observatory.Location
+import observatory.bearmug.ServiceLocator.Itr
+import org.openjdk.jmh.annotations._
+
+import scala.util.Random
+
+@State(Scope.Benchmark)
+@Warmup(iterations = 2, time = 5, timeUnit = TimeUnit.SECONDS)
+@Measurement(iterations = 5, time = 3, timeUnit = TimeUnit.SECONDS)
+@Fork(value = 1)
+class ServiceLocatorBenchmark {
+
+  val seqService = ServiceLocator.servePlain()
+
+  val parService = ServiceLocator.serveParallel()
+
+  val year = 2015
+  val resPath = s"${System.getProperty("buildDir")}/resources/test"
+  private val signature = Random.nextInt(10000000)
+  val stationsFile = s"/stations-jmh-$signature"
+  val temperaturesFile = s"/temperatures-jmh-$signature"
+  var preparedData: Itr = _
+
+  @Setup
+  def setup(): Unit = {
+    println(s"writing stations data to ${resPath + stationsFile}")
+    val stations = (1 to 10000).map(number => s"$number,$number,${Random.nextDouble()},${Random.nextDouble()}\n")
+    new PrintWriter(resPath + stationsFile) {
+      stations.foreach(write); flush(); close()
+    }
+    println(s"writing temperature data to ${resPath + temperaturesFile}")
+    val temperatures = (1 to 10000000).map(_ => {
+      val station = Random.nextInt(10000)
+      s"$station,$station,${Random.nextInt(11)},${1 + Random.nextInt(28)},${Random.nextInt(100) - 50.0}\n"
+    })
+    new PrintWriter(resPath + temperaturesFile) {
+      temperatures.foreach(write); flush(); close()
+    }
+
+    preparedData = ServiceLocator.servePlain().temperaturesOf(year, stationsFile, temperaturesFile)
+  }
+
+  @Benchmark
+  def plainTemperatures(): Iterable[(LocalDate, Location, Double)] =
+    seqService.temperaturesOf(year, stationsFile, temperaturesFile)
+
+  @Benchmark
+  def plainAvgTemperature(): Iterable[(Location, Double)] = seqService.yar(preparedData)
+
+  @Benchmark
+  def parallelTemperatures(): Iterable[(LocalDate, Location, Double)] =
+    parService.temperaturesOf(year, stationsFile, temperaturesFile)
+
+  @Benchmark
+  def parallelAvgTemperature(): Iterable[(Location, Double)] = parService.yar(preparedData)
+}

course-5/capstone/src/main/scala/observatory/bearmug/ServiceLocator.scala

@@ -13,6 +13,8 @@ sealed abstract class TemperatureService {
 
   def itr(year: Int, stnKey: String, stnSrc: BufferedSource, tmpSrc: BufferedSource): Itr
 
+  def stationsData(stnSrc: BufferedSource): Map[String, Location]
+
   final def temperaturesOf(year: Int, stationsFile: String, temperaturesFile: String): Itr =
     itr(
       year,
@@ -73,5 +75,55 @@ object ServiceLocator {
       }
   }
 
+  class ParallelService extends TemperatureService {
+
+    val chunkSize = 1024 * 32
+
+    def stationsData(stnSrc: BufferedSource): Map[String, Location] = stnSrc
+      .getLines()
+      .grouped(chunkSize)
+      .toIterable
+      .par
+      .flatMap(_.flatMap {
+        case stnPattern(stn, wban, latitude, longitude) =>
+          Option(s"$stn:$wban" -> Location(latitude.toDouble, longitude.toDouble))
+        case _ => None
+      }).toMap.seq
+
+    override def itr(
+                      year: Int,
+                      stnKey: String,
+                      stnSrc: BufferedSource,
+                      tmpSrc: BufferedSource): Itr = {
+
+      val stationsMap = stationsData(stnSrc)
+
+      tmpSrc
+        .getLines()
+        .grouped(chunkSize)
+        .toIterable.par
+        .flatMap(_.flatMap {
+          case tempPattern(stn, wban, month, day, tempF) => Some((
+            LocalDate.of(year, month.toInt, day.toInt),
+            stationsMap(s"$stn:$wban"),
+            Conversions.toCelsius(tempF.toDouble)))
+          case _ => None
+        }).seq
+    }
+
+    override def yar(itr: Itr): Iterable[(Location, Double)] = itr
+      .par
+      .groupBy(_._2)
+      .mapValues { values =>
+        values.aggregate((0.0, 0))(
+          (acc, data) => (acc._1 + data._3, acc._2 + 1),
+          (l, r) => (l._1 + r._1, l._2 + r._2)) match {
+          case (temp, count) => temp / count
+        }
+      }.seq
+  }
+
   def servePlain(): TemperatureService = new PlainService
+
+  def serveParallel(): TemperatureService = new ParallelService
 }

course-5/capstone/src/test/scala/observatory/bearmug/ServiceLocatorTest.scala

@@ -11,87 +11,88 @@ import org.scalatest.junit.JUnitRunner
 
 import scala.io.BufferedSource
 
-@RunWith(classOf[JUnitRunner])
 class ServiceLocatorTest extends FunSuite {
 
   def buff(s: String): BufferedSource = new BufferedSource(new ByteArrayInputStream(s.getBytes))
 
-  test("plainService picks stationsData well for proper format") {
-    assert(
-      new PlainService().stationsData(buff("007026,2212,+12.200,-020.010")) ==
-        Map("007026:2212" -> Location(+12.200, -020.010))
-    )
+  for (service <- List(ServiceLocator.servePlain(), ServiceLocator.serveParallel())) {
 
-    assert(
-      new PlainService().stationsData(buff(
-        """007026,2212,+12.200,-020.010
-          |1,2,+1.200,-01.010""".stripMargin)) ==
-        Map(
-          "007026:2212" -> Location(12.2, -20.01),
-          "1:2" -> Location(1.2, -1.01))
-    )
-  }
+    test(s"$service picks stationsData well for proper format") {
+      assert(
+        service.stationsData(buff("007026,2212,+12.200,-020.010")) ==
+          Map("007026:2212" -> Location(+12.200, -020.010))
+      )
 
-  test("plainService ignore stationsData well for wrong format") {
-    assert(new PlainService().stationsData(buff(",2212,+12.200,-020.010")) == Map.empty)
-    assert(new PlainService().stationsData(buff("007026,,+12.200,-020.010")) == Map.empty)
-    assert(new PlainService().stationsData(buff("007026,2212,12.200,-020.010")) == Map.empty)
-    assert(new PlainService().stationsData(buff("007026,2212,+12,-020.010")) == Map.empty)
-    assert(new PlainService().stationsData(buff("007026,2212,+12.,-020.010")) == Map.empty)
-    assert(new PlainService().stationsData(buff("007026,2212,,-020.010")) == Map.empty)
-    assert(new PlainService().stationsData(buff("007026,2212,+12.200,020.010")) == Map.empty)
-    assert(new PlainService().stationsData(buff("007026,2212,+12.200,")) == Map.empty)
-    assert(new PlainService().stationsData(buff("007026,2212,+12.200,20")) == Map.empty)
-  }
+      assert(
+        service.stationsData(buff(
+          """007026,2212,+12.200,-020.010
+            |1,2,+1.200,-01.010""".stripMargin)) ==
+          Map(
+            "007026:2212" -> Location(12.2, -20.01),
+            "1:2" -> Location(1.2, -1.01))
+      )
+    }
 
-  test("plainService picks temperatures for correct input") {
-    assert(new PlainService().itr(
-      2015,
-      "stations-source-file",
-      buff(
-        """007026,2212,+12.200,-020.010
-          |1,2,+1.200,-01.010""".stripMargin),
-      buff(
-        """007026,2212,07,11,+78.8
-          |007026,2212,07,12,68.8
-          |007026,2212,11,13,-56.2""".stripMargin)) ==
-      Seq(
-        (LocalDate.of(2015, 7, 11), Location(12.2, -20.01), 26.0),
-        (LocalDate.of(2015, 7, 12), Location(12.2, -20.01), 20.444444444444443),
-        (LocalDate.of(2015, 11, 13), Location(12.2, -20.01), -49.0)
+    test(s"$service ignore stationsData well for wrong format") {
+      assert(service.stationsData(buff(",2212,+12.200,-020.010")) == Map.empty)
+      assert(service.stationsData(buff("007026,,+12.200,-020.010")) == Map.empty)
+      assert(service.stationsData(buff("007026,2212,12.200,-020.010")) == Map.empty)
+      assert(service.stationsData(buff("007026,2212,+12,-020.010")) == Map.empty)
+      assert(service.stationsData(buff("007026,2212,+12.,-020.010")) == Map.empty)
+      assert(service.stationsData(buff("007026,2212,,-020.010")) == Map.empty)
+      assert(service.stationsData(buff("007026,2212,+12.200,020.010")) == Map.empty)
+      assert(service.stationsData(buff("007026,2212,+12.200,")) == Map.empty)
+      assert(service.stationsData(buff("007026,2212,+12.200,20")) == Map.empty)
+    }
+
+    test(s"$service picks temperatures for correct input") {
+      assert(service.itr(
+        2015,
+        "stations-source-file",
+        buff(
+          """007026,2212,+12.200,-020.010
+            |1,2,+1.200,-01.010""".stripMargin),
+        buff(
+          """007026,2212,07,11,+78.8
+            |007026,2212,07,12,68.8
+            |007026,2212,11,13,-56.2""".stripMargin)) ==
+        Seq(
+          (LocalDate.of(2015, 7, 11), Location(12.2, -20.01), 26.0),
+          (LocalDate.of(2015, 7, 12), Location(12.2, -20.01), 20.444444444444443),
+          (LocalDate.of(2015, 11, 13), Location(12.2, -20.01), -49.0)
+        )
       )
-    )
-  }
+    }
 
-  test("plainService omit temperatures for wrong input") {
-    assert(new PlainService().itr(2015, "stations-source-file",
-      buff("007026,2212,+12.200,-020.010"),
-      buff(
-        """,2212,07,11,78.8
-          |007026,,07,12,68.8
-          |007026,2212,,12,68.8
-          |007026,2212,07,,68.8
-          |007026,2212,07,12,
-          |007026,2212,07,12,68""".stripMargin)) ==
-      Seq.empty
-    )
-  }
+    test(s"$service omit temperatures for wrong input") {
+      assert(service.itr(2015, "stations-source-file",
+        buff("007026,2212,+12.200,-020.010"),
+        buff(
+          """,2212,07,11,78.8
+            |007026,,07,12,68.8
+            |007026,2212,,12,68.8
+            |007026,2212,07,,68.8
+            |007026,2212,07,12,
+            |007026,2212,07,12,68""".stripMargin)) ==
+        Seq.empty
+      )
+    }
 
-  test("plainService calc avg temperatures for correct input") {
-    val service = new PlainService()
-    assert(service.yar(service.itr(
-      2015,
-      "stations-source-file",
-      buff(
-        """007026,2212,+12.200,-020.010
-          |1,2,+1.200,-01.010""".stripMargin),
-      buff(
-        """007026,2212,07,11,+78.8
-          |007026,2212,07,12,8.8
-          |007026,2212,11,13,-56.2""".stripMargin))) ==
-      Map(
-        (Location(12.2,-20.01), -11.962962962962962)
+    test(s"$service calc avg temperatures for correct input") {
+      assert(service.yar(service.itr(
+        2015,
+        "stations-source-file",
+        buff(
+          """007026,2212,+12.200,-020.010
+            |1,2,+1.200,-01.010""".stripMargin),
+        buff(
+          """007026,2212,07,11,+78.8
+            |007026,2212,07,12,8.8
+            |007026,2212,11,13,-56.2""".stripMargin))) ==
+        Map(
+          (Location(12.2,-20.01), -11.962962962962962)
+        )
       )
-    )
+    }
   }
 }