[loganalysis] add cassandra integration

Author: ashrithr

.gitignore

@@ -2,6 +2,7 @@
 *.log
 wiki
 *.sc
+src/main/scala/com/cloudwick/Random.scala
 
 # sbt specific
 dist/*

README.md

@@ -0,0 +1,46 @@
+# Cloudwick Spark CodeBase
+
+This repository is a collection of Spark examples & use-case implementations for various components of the Spark eco-system including Spark-Core, Spark-Streaming, Spark-SQL, Spark-MLLib.
+
+## What does this repository contains ?
+
+* Spark core examples
+    * WordCount
+* Spark streaming examples
+    * NetworkWordCount
+    * NetworkWordCountWindowed
+    * RecoverableNetworkWordCount
+    * TwitterPopularTags
+    * KafkaWordCount
+* Spark core use-cases
+* Spark streaming use-cases
+    * LogAnalytics
+* Testing
+    * ScalaTest spec traits for Spark core, streaming and SQL API(s)
+    * Embedded Kafka and Zookeeper embedded server instances for testing
+    
+## How to download ?
+
+Simplest way is to clone the repository:
+
+```
+git clone https://github.com/cloudwicklabs/spark_codebase.git
+```
+    
+## How to run these ?
+
+To run any of these examples or use-cases you have to package them using a uber-jar (most of the examples depend of external dependencies, hence have to be packaged as a assembly jar).
+
+### Building an assembly jar
+
+From the project's home directory
+
+```
+sbt assembly
+```
+
+### Running using `spark-submit`
+
+[`spark-submit`](https://spark.apache.org/docs/latest/submitting-applications.html) is the simplest way to submit a spark application to the cluster and supports all the cluster manager's like stand-alone, yarn and mesos.
+
+Each of the main class has documentation on how to run it.

ReadMe.md

@@ -1,14 +1,25 @@
+import sbt.Keys._
+
 name := "spark_codebase"
 
 version := "1.0"
 
 scalaVersion := "2.10.5"
 
 resolvers ++= Seq(
-  "typesafe-repository" at "http://repo.typesafe.com/typesafe/releases"
+  "Typesafe repository snapshots"   at "http://repo.typesafe.com/typesafe/snapshots/",
+  "Typesafe repository releases"    at "http://repo.typesafe.com/typesafe/releases/",
+  "Sonatype repo"                   at "https://oss.sonatype.org/content/groups/scala-tools/",
+  "Sonatype releases"               at "https://oss.sonatype.org/content/repositories/releases",
+  "Sonatype snapshots"              at "https://oss.sonatype.org/content/repositories/snapshots",
+  "Sonatype staging"                at "http://oss.sonatype.org/content/repositories/staging",
+  "Java.net Maven2 Repository"      at "http://download.java.net/maven/2/",
+  "Twitter Repository"              at "http://maven.twttr.com",
+  "Websudos releases"               at "http://maven.websudos.co.uk/ext-release-local"
 )
 
 val sparkVersion = "1.2.1"
+val PhantomVersion = "1.6.0"
 
 libraryDependencies ++= Seq(
   "org.apache.spark" %% "spark-core" % sparkVersion % "provided"
@@ -18,19 +29,28 @@ libraryDependencies ++= Seq(
   "org.apache.spark" %% "spark-streaming-kafka" % sparkVersion
     exclude("org.apache.zookeeper", "zookeeper"),
   "org.apache.spark" %% "spark-streaming-twitter" % sparkVersion,
+  "org.slf4j" % "slf4j-api" % "1.7.12",
   "org.apache.kafka" %% "kafka" % "0.8.2.1"
     exclude("javax.jms", "jms")
     exclude("com.sun.jdmk", "jmxtools")
     exclude("com.sun.jmx", "jmxri")
-    exclude("org.slf4j", "slf4j-simple")
     exclude("log4j", "log4j")
     exclude("org.apache.zookeeper", "zookeeper")
-    exclude("com.101tec", "zkclient"),
-  "org.apache.curator" % "curator-test" % "2.4.0",
+    exclude("com.101tec", "zkclient")
+    excludeAll ExclusionRule(organization = "org.slf4j"),
+  "org.apache.curator" % "curator-test" % "2.4.0"
+    excludeAll ExclusionRule(organization = "io.netty")
+    excludeAll ExclusionRule(organization = "org.jboss.netty"),
   "com.101tec" % "zkclient" % "0.4"
     exclude("org.apache.zookeeper", "zookeeper"),
   "joda-time" % "joda-time" % "2.7",
   "com.maxmind.geoip2" % "geoip2" % "2.1.0",
+  "com.websudos" %% "phantom-dsl" % PhantomVersion,
+  "com.websudos" %% "phantom-zookeeper" % PhantomVersion
+    excludeAll ExclusionRule(organization = "io.netty")
+    excludeAll ExclusionRule(organization = "org.jboss.netty")
+    excludeAll ExclusionRule(organization = "org.slf4j"),
+  "com.typesafe" % "config" % "1.2.1",
   // Test dependencies
   "org.scalatest" %% "scalatest" % "2.2.4" % "test",
   "org.xerial.snappy" % "snappy-java" % "1.1.1.7"
@@ -44,4 +64,14 @@ parallelExecution in Test := false
 fork in Test := true
 
 // Skip running tests during assembly
-test in assembly := {}
+test in assembly := {}
+
+assemblyMergeStrategy in assembly := {
+  case PathList("javax", "servlet", xs @ _*)                => MergeStrategy.first
+  case PathList(ps @ _*) if ps.last endsWith ".html"        => MergeStrategy.first
+  case "application.conf"                                   => MergeStrategy.concat
+  case "com/twitter/common/args/apt/cmdline.arg.info.txt.1" => MergeStrategy.first
+  case x =>
+    val oldStrategy = (assemblyMergeStrategy in assembly).value
+    oldStrategy(x)
+}

build.sbt

@@ -1 +1,3 @@
-logLevel := Level.Warn
+logLevel := Level.Warn
+
+addSbtPlugin("net.virtual-void" % "sbt-dependency-graph" % "0.7.5")

project/plugins.sbt

@@ -0,0 +1,37 @@
+logs.dir=logs
+
+log4j.rootLogger=INFO, stdout
+
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c)%n
+
+log4j.appender.kafkaAppender=org.apache.log4j.DailyRollingFileAppender
+log4j.appender.kafkaAppender.DatePattern='.'yyyy-MM-dd-HH
+log4j.appender.kafkaAppender.File=${logs.dir}/server.log
+log4j.appender.kafkaAppender.layout=org.apache.log4j.PatternLayout
+log4j.appender.kafkaAppender.layout.ConversionPattern=[%d] %p %m (%c)%n
+
+log4j.appender.stateChangeAppender=org.apache.log4j.DailyRollingFileAppender
+log4j.appender.stateChangeAppender.DatePattern='.'yyyy-MM-dd-HH
+log4j.appender.stateChangeAppender.File=${logs.dir}/state-change.log
+log4j.appender.stateChangeAppender.layout=org.apache.log4j.PatternLayout
+log4j.appender.stateChangeAppender.layout.ConversionPattern=[%d] %p %m (%c)%n
+
+log4j.appender.requestAppender=org.apache.log4j.DailyRollingFileAppender
+log4j.appender.requestAppender.DatePattern='.'yyyy-MM-dd-HH
+log4j.appender.requestAppender.File=${logs.dir}/kafka-request.log
+log4j.appender.requestAppender.layout=org.apache.log4j.PatternLayout
+log4j.appender.requestAppender.layout.ConversionPattern=[%d] %p %m (%c)%n
+
+log4j.appender.cleanerAppender=org.apache.log4j.DailyRollingFileAppender
+log4j.appender.cleanerAppender.DatePattern='.'yyyy-MM-dd-HH
+log4j.appender.cleanerAppender.File=log-cleaner.log
+log4j.appender.cleanerAppender.layout=org.apache.log4j.PatternLayout
+log4j.appender.cleanerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n
+
+log4j.appender.controllerAppender=org.apache.log4j.DailyRollingFileAppender
+log4j.appender.controllerAppender.DatePattern='.'yyyy-MM-dd-HH
+log4j.appender.controllerAppender.File=${logs.dir}/controller.log
+log4j.appender.controllerAppender.layout=org.apache.log4j.PatternLayout
+log4j.appender.controllerAppender.layout.ConversionPattern=[%d] %p %m (%c)%n

src/main/resources/log4j.properties

@@ -0,0 +1,19 @@
+cassandra {
+  keyspace = "loganalytics"
+
+  host = "localhost"
+
+  # native_transport_port
+  #   port for the CQL native transport to listen for clients on
+  #
+  nativePort = 9042
+  # rpc_port
+  #   port for Thrift to listen for clients on
+  rpcPort = 9160
+
+  replication.strategy = "SimpleStrategy"
+  replication.factor = 3
+
+  # Size of pool used for cassandra operations
+  concurrency = 10
+}

src/main/resources/reference.conf

@@ -0,0 +1,26 @@
+package com.cloudwick.cassandra
+
+import com.cloudwick.cassandra.schema.{LocationVisitRecord, StatusCountRecord, LogVolumeRecord}
+import com.typesafe.config.ConfigFactory
+
+import scala.concurrent.Await
+import scala.concurrent.duration._
+
+/**
+ * Description goes here.
+ * @author ashrith
+ */
+trait Cassandra extends CassandraService {
+  private[this] val config = ConfigFactory.load()
+
+  protected val cassandraHost = config.getString("cassandra.host")
+  protected val nativePort = config.getInt("cassandra.nativePort")
+  protected val rpcPort = config.getInt("cassandra.rpcPort")
+
+  def installSchema(): Unit = {
+    Await.result(LogVolumeRecord.create.future(), 10.seconds)
+    Await.result(StatusCountRecord.create.future(), 10.seconds)
+    Await.result(LocationVisitRecord.create.future(), 10.seconds)
+  }
+}
+

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

@@ -0,0 +1,27 @@
+package com.cloudwick.cassandra
+
+import java.util.concurrent.Executors
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder
+import com.typesafe.config.ConfigFactory
+
+import scala.concurrent.ExecutionContext
+
+/**
+ * Description goes here.
+ * @author ashrith
+ */
+trait CassandraExecutionContext {
+  implicit val cassandraExecutionContext: ExecutionContext = CassandraExecutionContext.executionContext
+}
+
+object CassandraExecutionContext {
+  implicit val executionContext: ExecutionContext = {
+    val executor = {
+      val threadFactory = new ThreadFactoryBuilder().setNameFormat("cassandra-pool-%d").build()
+      val cassandraConcurrency = ConfigFactory.load().getInt("cassandra.concurrency")
+      Executors.newFixedThreadPool(cassandraConcurrency, threadFactory)
+    }
+    ExecutionContext.fromExecutor(executor)
+  }
+}

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

@@ -0,0 +1,40 @@
+package com.cloudwick.cassandra
+
+import com.cloudwick.cassandra.schema.{LocationVisit, LocationVisitRecord}
+import com.cloudwick.cassandra.service.LocationVisitServiceModule
+import com.cloudwick.logging.Logging
+import com.websudos.phantom.Implicits._
+
+import scala.concurrent.Future
+
+/**
+ * Description goes here.
+ * @author ashrith
+ */
+trait CassandraLocationVisitServiceModule extends LocationVisitServiceModule with CassandraService {
+
+  object locationVisitService extends LocationVisitService with Logging {
+
+    override def update(locationVisit: LocationVisit) = {
+      logger.trace(
+        s"Update location visit counter. Country: ${locationVisit.country} " +
+        s"City: ${locationVisit.city} " +
+        s"Count: ${locationVisit.totalCount}"
+      )
+
+      LocationVisitRecord.update
+        .where(_.country eqs locationVisit.country)
+        .and(_.city eqs locationVisit.city)
+        .modify(_.total_count increment locationVisit.totalCount)
+        .future()
+    }
+
+    def getCount(country: String, city: String): Future[Option[Long]] = {
+      LocationVisitRecord
+        .select(_.total_count)
+        .where(_.country eqs country)
+        .and(_.city eqs city)
+        .one()(session, cassandraExecutionContext)
+    }
+  }
+}

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

@@ -0,0 +1,30 @@
+package com.cloudwick.cassandra
+
+import com.cloudwick.cassandra.schema.{LogVolume, LogVolumeRecord}
+import com.cloudwick.cassandra.service.LogVolumeServiceModule
+import com.cloudwick.logging.Logging
+import com.websudos.phantom.Implicits._
+
+trait CassandraLogVolumeServiceModule extends LogVolumeServiceModule with CassandraService {
+
+  object logVolumeService extends LogVolumeService with Logging {
+    override def update(logVolume: LogVolume) = {
+      logger.trace(
+        s"Update volume per minute count. Minute: ${logVolume.timeStamp} " +
+        s"Count: ${logVolume.totalCount}"
+      )
+
+      LogVolumeRecord.update
+        .where(_.timeStamp eqs logVolume.timeStamp)
+        .modify(_.total_count increment logVolume.totalCount)
+        .future()
+    }
+
+    override def getCount(id: Long) = {
+      LogVolumeRecord
+        .select(_.total_count)
+        .where(_.timeStamp eqs id)
+        .one()(session, cassandraExecutionContext)
+    }
+  }
+}

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

@@ -0,0 +1,17 @@
+package com.cloudwick.cassandra
+
+import com.typesafe.config.ConfigFactory
+import com.websudos.phantom.zookeeper.{CassandraManager, SimpleCassandraConnector}
+
+/**
+ * Description goes here.
+ * @author ashrith
+ */
+trait CassandraService extends SimpleCassandraConnector with CassandraExecutionContext {
+   override def keySpace: String = {
+      val config = ConfigFactory.load()
+      config.getString("cassandra.keyspace")
+   }
+
+   override def manager: CassandraManager = ConfigurableCassandraManager
+}

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

@@ -0,0 +1,34 @@
+package com.cloudwick.cassandra
+
+import com.cloudwick.cassandra.schema.{StatusCount, StatusCountRecord}
+import com.cloudwick.cassandra.service.StatusCountServiceModule
+import com.cloudwick.logging.Logging
+import com.websudos.phantom.Implicits._
+
+import scala.concurrent.Future
+
+trait CassandraStatusCountServiceModule extends StatusCountServiceModule with CassandraService {
+
+  object statusCountService extends StatusCountService with Logging {
+
+    override def update(statusCount: StatusCount) = {
+      logger.trace(
+        s"Update status count counter. StatusCode: ${statusCount.statusCode} " +
+        s"Count: ${statusCount.totalCount}"
+      )
+
+      StatusCountRecord.update
+        .where(_.statusCode eqs statusCount.statusCode)
+        .modify(_.total_count increment statusCount.totalCount)
+        .future()
+    }
+
+    override def getCount(id: Int) = {
+      StatusCountRecord
+        .select(_.total_count)
+        .where(_.statusCode eqs id)
+        .one()(session, cassandraExecutionContext)
+    }
+  }
+
+}