Include KinesisWordCount example

Author: ashrithr

Vagrantfile

@@ -0,0 +1,41 @@
+# -*- mode: ruby -*-
+# vi: set ft=ruby :
+
+$script = <<SCRIPT
+cat << EOF >> /etc/hosts
+192.168.34.100 sparkmaster
+192.168.34.101 sparkworker1
+192.168.34.102 sparkworker2
+EOF
+
+sudo service iptables stop
+sudo chkconfig iptables stop
+sudo /usr/sbin/setenforce 0
+sudo sed -i.old s/SELINUX=enforcing/SELINUX=disabled/ /etc/selinux/config
+SCRIPT
+
+boxes = [
+    { :name => :sparkmaster,  :ip => '192.168.34.100', :cpus => 2, :memory => 1024 },
+    { :name => :sparkworker1, :ip => '192.168.34.101', :cpus => 2, :memory => 2048 },
+    { :name => :sparkworker2, :ip => '192.168.34.102', :cpus => 2, :memory => 2048 },
+]
+
+VAGRANT_API_VERSION = "2"
+
+Vagrant.configure(VAGRANT_API_VERSION) do |conf|
+  conf.vm.box = "chef/centos-6.5"
+
+  boxes.each do |box|
+    conf.vm.define box[:name] do |config|
+      config.vm.network 'private_network', ip: box[:ip]
+      config.vm.hostname = box[:name].to_s
+      config.vm.provider "virtualbox" do |v|
+        v.customize ["modifyvm", :id, "--memory", box[:memory]]
+        v.customize ["modifyvm", :id, "--cpus", box[:cpus]]
+      end
+
+      #provisioning
+      config.vm.provision :shell, inline: $script
+    end
+  end
+end

build.sbt

@@ -1,3 +1,4 @@
+import sbt.ExclusionRule
 import sbt.Keys._
 
 name := "spark_codebase"
@@ -18,7 +19,7 @@ resolvers ++= Seq(
   "Websudos releases"               at "http://maven.websudos.co.uk/ext-release-local"
 )
 
-val sparkVersion = "1.2.1"
+val sparkVersion = "1.3.0"
 val PhantomVersion = "1.6.0"
 
 libraryDependencies ++= Seq(
@@ -29,6 +30,8 @@ libraryDependencies ++= Seq(
   "org.apache.spark" %% "spark-streaming-kafka" % sparkVersion
     exclude("org.apache.zookeeper", "zookeeper"),
   "org.apache.spark" %% "spark-streaming-twitter" % sparkVersion,
+  "org.apache.spark" %% "spark-streaming-kinesis-asl" % sparkVersion
+    excludeAll ExclusionRule(organization = "org.apache.spark", name = "spark-streaming_2.10"),
   "org.slf4j" % "slf4j-api" % "1.7.12",
   "org.apache.kafka" %% "kafka" % "0.8.2.1"
     exclude("javax.jms", "jms")
@@ -45,7 +48,8 @@ libraryDependencies ++= Seq(
   "com.101tec" % "zkclient" % "0.4"
     exclude("org.apache.zookeeper", "zookeeper"),
   "joda-time" % "joda-time" % "2.7",
-  "com.maxmind.geoip2" % "geoip2" % "2.1.0",
+  "com.maxmind.geoip2" % "geoip2" % "2.1.0"
+    exclude("org.apache.httpcomponents", "httpclient"),
   "com.websudos" %% "phantom-dsl" % PhantomVersion
     exclude("com.google.guava", "guava"),
   "com.websudos" %% "phantom-zookeeper" % PhantomVersion
@@ -73,9 +77,20 @@ test in assembly := {}
 assemblyMergeStrategy in assembly := {
   case PathList("javax", "servlet", xs @ _*)                => MergeStrategy.first
   case PathList(ps @ _*) if ps.last endsWith ".html"        => MergeStrategy.first
+  case PathList("com", "google", "common", "base", xs @ _*) => MergeStrategy.first
   case "application.conf"                                   => MergeStrategy.concat
   case "com/twitter/common/args/apt/cmdline.arg.info.txt.1" => MergeStrategy.first
+  case "org/apache/spark/unused/UnusedStubClass.class"      => MergeStrategy.first
+  case "log4j.properties"                                   => MergeStrategy.first
+  case "reference.conf"                                     => MergeStrategy.concat
   case x =>
     val oldStrategy = (assemblyMergeStrategy in assembly).value
     oldStrategy(x)
+}
+
+assemblyExcludedJars in assembly <<= (fullClasspath in assembly) map { cp =>
+  val excludes = Set(
+    "commons-httpclient-3.1.jar"
+  )
+  cp filter { jar => excludes(jar.data.getName) }
 }

src/main/resources/log4j.properties

@@ -1,6 +1,6 @@
 logs.dir=logs
 
-log4j.rootLogger=INFO, stdout
+log4j.rootLogger=WARN, stdout
 
 log4j.appender.stdout=org.apache.log4j.ConsoleAppender
 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout

src/main/scala/com/cloudwick/cassandra/CassandraLocationVisitServiceModule.scala

@@ -2,7 +2,7 @@ package com.cloudwick.cassandra
 
 import com.cloudwick.cassandra.schema.{LocationVisit, LocationVisitRecord}
 import com.cloudwick.cassandra.service.LocationVisitServiceModule
-import com.cloudwick.logging.Logging
+import com.cloudwick.logging.LazyLogging
 import com.websudos.phantom.Implicits._
 
 import scala.concurrent.Future
@@ -13,7 +13,7 @@ import scala.concurrent.Future
  */
 trait CassandraLocationVisitServiceModule extends LocationVisitServiceModule with CassandraService {
 
-  object locationVisitService extends LocationVisitService with Logging {
+  object locationVisitService extends LocationVisitService with LazyLogging {
 
     override def update(locationVisit: LocationVisit) = {
       logger.trace(

src/main/scala/com/cloudwick/cassandra/CassandraLogVolumeServiceModule.scala

@@ -2,12 +2,12 @@ package com.cloudwick.cassandra
 
 import com.cloudwick.cassandra.schema.{LogVolume, LogVolumeRecord}
 import com.cloudwick.cassandra.service.LogVolumeServiceModule
-import com.cloudwick.logging.Logging
+import com.cloudwick.logging.LazyLogging
 import com.websudos.phantom.Implicits._
 
 trait CassandraLogVolumeServiceModule extends LogVolumeServiceModule with CassandraService {
 
-  object logVolumeService extends LogVolumeService with Logging {
+  object logVolumeService extends LogVolumeService with LazyLogging {
     override def update(logVolume: LogVolume) = {
       logger.trace(
         s"Update volume per minute count. Minute: ${logVolume.timeStamp} " +

src/main/scala/com/cloudwick/cassandra/CassandraStatusCountServiceModule.scala

@@ -2,14 +2,12 @@ package com.cloudwick.cassandra
 
 import com.cloudwick.cassandra.schema.{StatusCount, StatusCountRecord}
 import com.cloudwick.cassandra.service.StatusCountServiceModule
-import com.cloudwick.logging.Logging
+import com.cloudwick.logging.LazyLogging
 import com.websudos.phantom.Implicits._
 
-import scala.concurrent.Future
-
 trait CassandraStatusCountServiceModule extends StatusCountServiceModule with CassandraService {
 
-  object statusCountService extends StatusCountService with Logging {
+  object statusCountService extends StatusCountService with LazyLogging {
 
     override def update(statusCount: StatusCount) = {
       logger.trace(

src/main/scala/com/cloudwick/logging/Logging.scala

@@ -2,6 +2,6 @@ package com.cloudwick.logging
 
 import org.slf4j.{LoggerFactory, Logger}
 
-trait Logging {
+trait LazyLogging {
   protected lazy val logger: Logger = LoggerFactory.getLogger(getClass.getName)
 }

src/main/scala/com/cloudwick/spark/embedded/KafkaServer.scala

@@ -3,7 +3,7 @@ package com.cloudwick.spark.embedded
 import java.nio.file.Files
 import java.util.Properties
 
-import com.cloudwick.logging.Logging
+import com.cloudwick.logging.LazyLogging
 import kafka.admin.AdminUtils
 import kafka.server.{KafkaConfig, KafkaServerStartable}
 import kafka.utils.ZKStringSerializer
@@ -21,7 +21,7 @@ import scala.concurrent.duration._
  * @param config Broker configuration settings.  Used to modify, for example, on which port the
  *               broker should listen to.
  */
-class KafkaServer(config: Properties = new Properties) extends Logging {
+class KafkaServer(config: Properties = new Properties) extends LazyLogging {
   private val defaultZkConnect = "127.0.0.1:2181"
   private val logDir = Files.createTempDirectory(this.getClass.getSimpleName)
 

src/main/scala/com/cloudwick/spark/embedded/ZookeeperServer.scala

@@ -1,12 +1,12 @@
 package com.cloudwick.spark.embedded
 
-import com.cloudwick.logging.Logging
+import com.cloudwick.logging.LazyLogging
 import org.apache.curator.test.TestingServer
 
 /**
  * Runs an in-memory, "embedded" instance of a ZooKeeper server.
  */
-class ZookeeperServer(val port: Int = 2181) extends Logging {
+class ZookeeperServer(val port: Int = 2181) extends LazyLogging {
   logger.debug(s"Starting embedded ZooKeeper server on port $port...")
 
   // Creates a new instance of zookeeper server when an instance of this class is created

src/main/scala/com/cloudwick/spark/examples/core/WordCountRunner.scala

@@ -1,11 +1,12 @@
 package com.cloudwick.spark.examples.core
 
+import com.cloudwick.logging.LazyLogging
 import org.apache.spark.{Logging, SparkConf, SparkContext}
 
 /**
  * Simple word count program to illustrate spark standalone applications usage
  */
-object WordCountRunner extends App with Logging {
+object WordCountRunner extends App with LazyLogging {
   if (args.length < 2) {
     System.err.println("Usage: WordCountRunner input_path output_path")
     System.exit(1)
@@ -20,7 +21,7 @@ object WordCountRunner extends App with Logging {
   val lines = sc.textFile(inputPath)
   val counts = WordCount.count(lines, stopWords)
 
-  // log.info(counts.collect().mkString("[", ", ", "]"))
+  // logger.info(counts.collect().mkString("[", ", ", "]"))
 
   counts.saveAsTextFile(outputPath)
 }

src/main/scala/com/cloudwick/spark/examples/streaming/kafka/KafkaWordCount.scala

@@ -2,13 +2,13 @@ package com.cloudwick.spark.examples.streaming.kafka
 
 import java.nio.file.Files
 
+import com.cloudwick.logging.LazyLogging
 import com.cloudwick.spark.examples.core.WordCount
 import com.cloudwick.spark.examples.streaming.local.NetworkWordCountWindowed
-import com.cloudwick.spark.examples.streaming.util.Utils
 import org.apache.spark.rdd.RDD
 import org.apache.spark.streaming.kafka.KafkaUtils
 import org.apache.spark.streaming.{Seconds, StreamingContext, Time}
-import org.apache.spark.{Logging, SparkConf}
+import org.apache.spark.SparkConf
 
 /**
  * Consumes messages from one or more topics in Kafka and does word-count.
@@ -30,9 +30,9 @@ import org.apache.spark.{Logging, SparkConf}
  * 7. Check the offset consumption of the topic
  *      `bin/kafka-consumer-offset-checker.sh --zookeeper localhost:2181 --topic test-wc --group stcg`
  */
-object KafkaWordCount extends App with Logging {
+object KafkaWordCount extends App with LazyLogging {
   if (args.length < 4) {
-    log.error(
+    logger.error(
       """
         |Usage: KafkaWordCount <zkQuorum> <group> <topics> <numThreads>
         |         zkQuorum - Zookeeper quorum (hostname:port,hostname:port,..)
@@ -45,8 +45,6 @@ object KafkaWordCount extends App with Logging {
     System.exit(1)
   }
 
-  Utils.setSparkLogLevels()
-
   val Array(zkQuorum, group, topics, numThreads) = args
   val batchDuration = Seconds(5)
   val windowDuration = Seconds(30)

src/main/scala/com/cloudwick/spark/examples/streaming/kinesis/KinesisWordCount.scala

@@ -0,0 +1,80 @@
+package com.cloudwick.spark.examples.streaming.kinesis
+
+import com.amazonaws.auth.AWSCredentials
+import com.amazonaws.services.kinesis.AmazonKinesisClient
+import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream
+import com.cloudwick.logging.LazyLogging
+import org.apache.spark.storage.StorageLevel
+import org.apache.spark.streaming.StreamingContext._
+import org.apache.spark.streaming.kinesis.KinesisUtils
+import org.apache.spark.streaming.{Seconds, StreamingContext}
+import org.apache.spark.SparkConf
+
+/**
+ * Kinesis word count example
+ *
+ * Running this example locally:
+ *
+ *  `spark-submit --class com.cloudwick.spark.examples.streaming.kinesis.KinesisWordCount
+ *    --master "local[*]" target/scala-2.10/spark_codebase-assembly-1.0.jar
+ *    <stream-name> <aws-access-key> <aws-secret-key> <endpoint-url>`
+ */
+object KinesisWordCount extends App with LazyLogging {
+  if (args.length < 4) {
+    logger.error(
+      """
+        |Usage: KinesisWordCount <stream-name> <aws-access-key> <aws-secret-key> <endpoint-url>
+        |         stream-name - is the name of the kinesis stream
+        |         aws-access-key - is the aws access key
+        |         aws-secret-key - is the aws secret access keuy
+        |         endpoint-url - is the endpoint of the kinesis service
+      """.stripMargin
+    )
+    System.exit(1)
+  }
+
+  def fromCredentials(awsAccessKey: String,
+                      awsSecretKey: String,
+                      awsEndPoint: String): AmazonKinesisClient = {
+    val credentials = new AWSCredentials {
+      override def getAWSAccessKeyId: String = awsAccessKey
+
+      override def getAWSSecretKey: String = awsSecretKey
+    }
+    val client = new AmazonKinesisClient(credentials)
+    client.setEndpoint(awsEndPoint)
+    client
+  }
+
+  val Array(streamName, awsAccessKey, awsSecretKey, endPointUrl) = args
+
+  // Determine the number of shards for a specified stream, so that we could create one kinesis
+  // receiver for each shard
+  val kinesisClient = fromCredentials(awsAccessKey, awsSecretKey, endPointUrl)
+  val numShards = kinesisClient.describeStream(streamName).getStreamDescription.getShards.size
+
+  val batchDuration = Seconds(2)
+  val sparkConf = new SparkConf().setAppName("KinesisWordCount").setMaster("local[*]")
+  val ssc = new StreamingContext(sparkConf, batchDuration)
+
+  // create receivers
+  // set aws.accessKeyId and aws.secretKey as system properties
+  System.setProperty("aws.accessKeyId", awsAccessKey)
+  System.setProperty("aws.secretKey", awsSecretKey)
+  val kinesisStreams = (0 until numShards).map { i =>
+    KinesisUtils.createStream(ssc, streamName, endPointUrl, batchDuration,
+      InitialPositionInStream.TRIM_HORIZON, StorageLevel.MEMORY_AND_DISK_2)
+  }
+
+  // union all the streams
+  val unionStream = ssc.union(kinesisStreams)
+
+  // convert each record of type Byte to string
+  val words = unionStream.flatMap(new String(_).split("\\s+"))
+  val wordCounts = words.map(word => (word, 1)).reduceByKey(_ + _)
+
+  wordCounts.print()
+
+  ssc.start()
+  ssc.awaitTermination()
+}

src/main/scala/com/cloudwick/spark/examples/streaming/local/NetworkWordCountRunner.scala

@@ -1,7 +1,6 @@
 package com.cloudwick.spark.examples.streaming.local
 
 import com.cloudwick.spark.examples.core.WordCount
-import com.cloudwick.spark.examples.streaming.util.Utils
 import org.apache.spark.rdd.RDD
 import org.apache.spark.storage.StorageLevel
 import org.apache.spark.streaming.{Seconds, StreamingContext, Time}
@@ -31,8 +30,6 @@ object NetworkWordCountRunner extends App with Logging {
   val Array(host, port, batchInterval) = args
   val stopWords = Set("a", "an", "the")
 
-  Utils.setSparkLogLevels()
-
   val conf = new SparkConf().setAppName("NetworkWordCount")
   val ssc = new StreamingContext(conf, Seconds(batchInterval.toInt))
 

src/main/scala/com/cloudwick/spark/examples/streaming/local/NetworkWordCountWindowedRunner.scala

@@ -3,7 +3,6 @@ package com.cloudwick.spark.examples.streaming.local
 import java.nio.file.Files
 import com.cloudwick.spark.examples.core.WordCount
 import com.cloudwick.spark.examples.streaming.local.NetworkWordCountWindowed._
-import com.cloudwick.spark.examples.streaming.util.Utils
 import org.apache.spark.rdd.RDD
 import org.apache.spark.{Logging, SparkConf}
 import org.apache.spark.storage.StorageLevel
@@ -40,8 +39,6 @@ object NetworkWordCountWindowedRunner extends App with Logging {
 
   log.info(s"Connecting to host: $hostname port: $port")
 
-  Utils.setSparkLogLevels()
-
   // Create a local StreamingContext with master & specified batch interval
   val conf = new SparkConf().setAppName("NetworkWordCount")
   val ssc = new StreamingContext(conf, Seconds(5))

src/main/scala/com/cloudwick/spark/examples/streaming/twitter/TwitterPopularTags.scala

@@ -1,6 +1,5 @@
 package com.cloudwick.spark.examples.streaming.twitter
 
-import com.cloudwick.spark.examples.streaming.util.Utils
 import org.apache.spark.streaming.twitter.TwitterUtils
 import org.apache.spark.streaming.{Seconds, StreamingContext}
 import org.apache.spark.{SparkConf, Logging}
@@ -23,8 +22,6 @@ object TwitterPopularTags extends App with Logging {
     System.exit(1)
   }
 
-  Utils.setSparkLogLevels()
-
   val Array(consumerKey, consumerSecret, accessToken, accessTokenSecret) = args.take(4)
   val filters = args.takeRight(args.length - 4)