adding streaming example and providing setup instructions and tests

Author: Katrin Shechtman

README.md

@@ -30,7 +30,7 @@ You should see `scala>` after some more updates (Use `:q` for quitting). You're
 
 # Different ways to execute the code
 
-`sbt test` - executes all tests
+`sbt "testOnly sample.<name-of-the-test>""` - executes specific test
 
 `sbt run` - executing Main with Spark in-memory
 

build.sbt

@@ -13,7 +13,10 @@ libraryDependencies ++= {
       ExclusionRule(organization = "org.scalactic"),
       ExclusionRule(organization = "org.scalatest")
     ),
-    "org.apache.spark" %% "spark-hive" % sparkVersion
+    "org.apache.spark" %% "spark-hive" % sparkVersion,
+    "org.apache.spark" % "spark-streaming_2.11" % sparkVersion % "provided",
+    "com.typesafe" % "config" % "1.3.1",
+    "org.jfarcand" % "wcs" % "1.5"
   )
 }
 
@@ -25,9 +28,11 @@ assemblyMergeStrategy in assembly := {
 fork in run := true
 javaOptions in run ++= Seq(
   "-Dlog4j.configuration=log4j.properties")
-fork in Test := true
+fork in Test := false
 javaOptions in Test ++= Seq(
   "-Dlog4j.configuration=log4j.properties")
 
+parallelExecution in Test := false
+
 run in Compile <<= Defaults.runTask(fullClasspath in Compile, mainClass in(Compile, run), runner in(Compile, run))
 runMain in Compile <<= Defaults.runMainTask(fullClasspath in Compile, runner in(Compile, run))

src/main/resources/application.conf

@@ -0,0 +1,3 @@
+// NOTE: it assumes these jars are present at this location in Spark executors.
+//spark.jars.location=["local:///usr/local/spark-2.0.1-bin-hadoop2.7/mysql-connector-java-6.0.5.jar"]
+spark.master="local[*]"// "spark://127.0.0.1:7077"

src/main/scala/sample/CryptoCurrencyFeedReceiver.scala

@@ -0,0 +1,32 @@
+package sample
+
+import org.apache.spark.storage.StorageLevel
+import org.apache.spark.streaming.receiver.Receiver
+import org.jfarcand.wcs.{TextListener, WebSocket}
+
+
+class CryptoCurrencyFeedReceiver extends Receiver[String](StorageLevel.MEMORY_ONLY) with Runnable {
+
+  private[this] lazy val thread: Thread = new Thread(this)
+
+  private[this] lazy val webSocket: WebSocket =
+    WebSocket().open("wss://stream.binance.com:9443/ws/ethbtc@aggTrade")
+
+  override def onStart() = {
+    thread.start()
+  }
+
+  override def onStop() = {
+    webSocket.close
+    thread.interrupt()
+  }
+
+  override def run() = {
+    webSocket.listener(new TextListener {
+      override def onMessage(message: String) {
+        println(message)
+        store(message)
+      }
+    })
+  }
+}

src/main/scala/Main.scala src/main/scala/sample/Main.scala

@@ -1,7 +1,8 @@
+package sample
 
 import java.io.File
 
-import org.apache.spark._
+import org.apache.spark.{SparkConf, SparkContext}
 
 object Main extends App {
 

src/main/scala/sample/Utils.scala

@@ -0,0 +1,33 @@
+package sample
+
+import com.typesafe.config.Config
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.streaming.{Duration, StreamingContext}
+
+object Utils {
+
+  def getOrCreateSparkSession(conf: Config) = {
+
+    val appName = "Scala for Spark"
+    val master = conf.getString("spark.master")
+    // val jars = conf.getString("spark.jars.location")
+    val session = SparkSession.builder
+      .master(master)
+      .appName(appName)
+      // .config("spark.jars", jars)
+      .getOrCreate()
+    session
+  }
+
+  def createSparkStreamingContext(conf: Config, time: Duration) = {
+
+    val appName = "Scala for Spark"
+    val master = conf.getString("spark.master")
+    val sparkConf = new SparkConf()
+      .setMaster(master)
+      .setAppName(appName)
+
+    new StreamingContext(sparkConf, time)
+  }
+}

src/test/resources/application.conf

@@ -0,0 +1,7 @@
+db.default.port=3307
+db.default.driver="com.mysql.cj.jdbc.Driver"
+db.default.url="jdbc:mysql://127.0.0.1:"${db.default.port}"/test"
+db.default.user="root"
+db.default.password=""
+spark.jars.location=""
+spark.master="local[*]"

src/test/scala/sample/LocalStreamingSparkSpec.scala

@@ -0,0 +1,47 @@
+package sample
+
+import com.typesafe.config.ConfigFactory
+import org.apache.spark.streaming.Seconds
+import org.scalatest.concurrent.Eventually
+import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}
+
+import scala.collection.mutable
+
+class HelloLocalSparkSpec extends WordSpecLike
+  with Matchers
+  with BeforeAndAfterAll
+  with Eventually {
+
+  val conf = ConfigFactory.load()
+
+  lazy val ssc = Utils.createSparkStreamingContext(conf, Seconds(10))
+
+  override protected def afterAll(): Unit = {
+    super.afterAll()
+    ssc.stop()
+  }
+
+  "Hello Spark Streaming" should {
+
+    "successfully receives multiple batches" in {
+
+      val items = mutable.Map.empty[Int, Long].withDefaultValue(0L)
+
+      val stream = ssc.receiverStream(new CryptoCurrencyFeedReceiver())
+
+      stream.foreachRDD(rdd => items += rdd.id -> (items(rdd.id) + rdd.count()))
+
+      ssc.start()
+
+      Thread.sleep(22000)
+
+      println(items)
+
+      assert(items.exists {
+        case (_, trades) => trades > 0
+      })
+
+    }
+  }
+}
+

src/test/scala/WordCountTest.scala src/test/scala/sample/WordCountTest.scala

@@ -1,15 +1,15 @@
+package sample
+
+import org.apache.spark.{SparkConf, SparkContext}
 import org.scalatest.{BeforeAndAfterAll, FunSuite}
-import org.apache.spark.SparkContext
-import org.apache.spark.SparkConf
 
 class WordCountTest extends FunSuite with BeforeAndAfterAll {
 
-  private var sparkConf: SparkConf = _
-  private var sc: SparkContext = _
+  lazy val sparkConf: SparkConf = new SparkConf().setAppName("unit-testing").setMaster("local")
+  lazy val sc: SparkContext = new SparkContext(sparkConf)
 
   override def beforeAll() {
-    sparkConf = new SparkConf().setAppName("unit-testing").setMaster("local")
-    sc = new SparkContext(sparkConf)
+
   }
 
   test("get word count rdd") {

src/test/scala/WordCountTestWithTestingBase.scala src/test/scala/sample/WordCountTestWithTestingBase.scala

@@ -1,3 +1,5 @@
+package sample
+
 import com.holdenkarau.spark.testing.SharedSparkContext
 import org.scalatest.FunSuite