Publish AnalyzedItems to Cassandra

README.md

@@ -17,6 +17,9 @@ Run it via:
 # set up all the requisite environment variables
 export FORTIS_APPINSIGHTS_IKEY="..."
 export FORTIS_FEATURE_SERVICE_HOST="..."
+export FORTIS_CASSANDRA_HOST="..."
+export FORTIS_CASSANDRA_USER="..."
+export FORTIS_CASSANDRA_PASSWORD="..."
 export INSTAGRAM_AUTH_TOKEN="..."
 export OXFORD_VISION_TOKEN="..."
 export OXFORD_SPEECH_TOKEN="..."

build.sbt

@@ -17,6 +17,7 @@ val sparkVersion = "2.1.0"
 // Dependencies provided by the Spark distro
 libraryDependencies ++= Seq(
   "org.apache.spark" %% "spark-core" % sparkVersion,
+  "org.apache.spark" %% "spark-sql" % sparkVersion,
   "org.apache.spark" %% "spark-streaming" % sparkVersion
 ).map(_ % "provided")
 
@@ -33,6 +34,7 @@ libraryDependencies ++= Seq(
   "org.apache.commons" % "commons-collections4" % "4.1",
   "com.microsoft.azure" %% "spark-streaming-eventhubs" % "2.0.5",
   "com.esotericsoftware.kryo" % "kryo" % "2.24.0",
+  "com.datastax.spark" %% "spark-cassandra-connector" % "2.0.2",
   "net.liftweb" %% "lift-json" % "3.0.1",
   "org.scalaj" %% "scalaj-http" % "2.3.0",
   "net.lingala.zip4j" % "zip4j" % "1.3.2",

src/main/scala/com/microsoft/partnercatalyst/fortis/spark/ProjectFortis.scala

@@ -2,6 +2,7 @@ package com.microsoft.partnercatalyst.fortis.spark
 
 import com.microsoft.partnercatalyst.fortis.spark.logging.AppInsights
 import com.microsoft.partnercatalyst.fortis.spark.pipeline._
+import com.microsoft.partnercatalyst.fortis.spark.sinks.cassandra.CassandraSink
 import com.microsoft.partnercatalyst.fortis.spark.streamprovider.ConnectorConfig
 import com.microsoft.partnercatalyst.fortis.spark.transforms.image.{ImageAnalysisAuth, ImageAnalyzer}
 import com.microsoft.partnercatalyst.fortis.spark.transforms.language.{LanguageDetector, LanguageDetectorAuth}
@@ -88,12 +89,8 @@ object ProjectFortis extends App {
 
     // Attach each pipeline (aka code path)
     // 'fortisEvents' is the stream of analyzed data aggregated (union) from all pipelines
-    val fortisEvents = pipelines.flatMap(
-      pipeline => pipeline(streamProvider, streamRegistry, ssc, TransformContext)
-    ).reduceOption(_.union(_))
-
-    // TODO: other computations and save to DB
-    fortisEvents.foreach(_.print())
+    val fortisEvents = pipelines.flatMap(pipeline => pipeline(streamProvider, streamRegistry, ssc, TransformContext)).reduceOption(_.union(_))
+    CassandraSink(fortisEvents, "fortistest", "events") // todo: fill in real values
 
     ssc.checkpoint(Settings.progressDir)
     ssc

src/main/scala/com/microsoft/partnercatalyst/fortis/spark/dto/AnalyzedItem.scala

@@ -1,6 +1,9 @@
 package com.microsoft.partnercatalyst.fortis.spark.dto
 
+import java.util.UUID
+
 case class AnalyzedItem(
+  id: UUID,
   createdAtEpoch: Long,
   body: String,
   title: String,

src/main/scala/com/microsoft/partnercatalyst/fortis/spark/pipeline/BingPipeline.scala

@@ -1,6 +1,7 @@
 package com.microsoft.partnercatalyst.fortis.spark.pipeline
 
 import java.time.Instant.now
+import java.util.UUID.randomUUID
 
 import com.github.catalystcode.fortis.spark.streaming.bing.dto.BingPost
 import com.microsoft.partnercatalyst.fortis.spark.dto.{Analysis, AnalyzedItem}
@@ -17,6 +18,7 @@ object BingPipeline extends Pipeline {
 
   private def convertToSchema(stream: DStream[BingPost], transformContext: TransformContext): DStream[AnalyzedItem] = {
     stream.map(post => AnalyzedItem(
+      id = randomUUID(),
       createdAtEpoch = now.getEpochSecond,
       body = post.snippet,
       title = post.name,

src/main/scala/com/microsoft/partnercatalyst/fortis/spark/pipeline/FacebookPipeline.scala

@@ -1,6 +1,7 @@
 package com.microsoft.partnercatalyst.fortis.spark.pipeline
 
 import java.time.Instant.now
+import java.util.UUID.randomUUID
 
 import com.github.catalystcode.fortis.spark.streaming.facebook.dto.FacebookPost
 import com.microsoft.partnercatalyst.fortis.spark.dto.{Analysis, AnalyzedItem}
@@ -19,6 +20,7 @@ object FacebookPipeline extends Pipeline {
     import transformContext._
 
     stream.map(post => AnalyzedItem(
+      id = randomUUID(),
       createdAtEpoch = now.getEpochSecond,
       body = post.post.getMessage,
       title = "",

src/main/scala/com/microsoft/partnercatalyst/fortis/spark/pipeline/InstagramPipeline.scala

@@ -1,6 +1,7 @@
 package com.microsoft.partnercatalyst.fortis.spark.pipeline
 
 import java.time.Instant.now
+import java.util.UUID.randomUUID
 
 import com.github.catalystcode.fortis.spark.streaming.instagram.dto.InstagramItem
 import com.microsoft.partnercatalyst.fortis.spark.dto.AnalyzedItem
@@ -18,6 +19,7 @@ object InstagramPipeline extends Pipeline {
         // do computer vision analysis
         val analysis = imageAnalyzer.analyze(instagram.images.standard_resolution.url)
         AnalyzedItem(
+          id = randomUUID(),
           createdAtEpoch = now.getEpochSecond,
           body = analysis.summary.getOrElse(""),
           title = instagram.caption.text,

src/main/scala/com/microsoft/partnercatalyst/fortis/spark/pipeline/RadioPipeline.scala

@@ -1,6 +1,7 @@
 package com.microsoft.partnercatalyst.fortis.spark.pipeline
 
 import java.time.Instant.now
+import java.util.UUID.randomUUID
 
 import com.microsoft.partnercatalyst.fortis.spark.dto.{Analysis, AnalyzedItem}
 import com.microsoft.partnercatalyst.fortis.spark.streamprovider.{ConnectorConfig, StreamProvider}
@@ -17,6 +18,7 @@ object RadioPipeline extends Pipeline {
 
   private def convertToSchema(stream: DStream[RadioTranscription], transformContext: TransformContext): DStream[AnalyzedItem] = {
     stream.map(transcription => AnalyzedItem(
+      id = randomUUID(),
       createdAtEpoch = now.getEpochSecond,
       body = transcription.text,
       title = "",

src/main/scala/com/microsoft/partnercatalyst/fortis/spark/pipeline/TadawebPipeline.scala

@@ -1,6 +1,7 @@
 package com.microsoft.partnercatalyst.fortis.spark.pipeline
 
 import java.time.Instant.now
+import java.util.UUID.randomUUID
 
 import com.microsoft.partnercatalyst.fortis.spark.dto.{Analysis, AnalyzedItem}
 import com.microsoft.partnercatalyst.fortis.spark.streamprovider.{ConnectorConfig, StreamProvider}
@@ -19,6 +20,7 @@ object TadawebPipeline extends Pipeline {
     import transformContext._
 
     stream.map(tada => AnalyzedItem(
+      id = randomUUID(),
       createdAtEpoch = now.getEpochSecond,
       body = tada.text,
       title = tada.title,

src/main/scala/com/microsoft/partnercatalyst/fortis/spark/pipeline/TwitterPipeline.scala

@@ -1,6 +1,7 @@
 package com.microsoft.partnercatalyst.fortis.spark.pipeline
 
 import java.time.Instant.now
+import java.util.UUID.randomUUID
 
 import com.microsoft.partnercatalyst.fortis.spark.dto.{Analysis, AnalyzedItem}
 import com.microsoft.partnercatalyst.fortis.spark.streamprovider.{ConnectorConfig, StreamProvider}
@@ -19,6 +20,7 @@ object TwitterPipeline extends Pipeline {
     import transformContext._
 
     stream.map(tweet => AnalyzedItem(
+      id = randomUUID(),
       createdAtEpoch = now.getEpochSecond,
       body = tweet.getText,
       title = "",

src/main/scala/com/microsoft/partnercatalyst/fortis/spark/sinks/cassandra/CassandraConfig.scala

@@ -0,0 +1,15 @@
+package com.microsoft.partnercatalyst.fortis.spark.sinks.cassandra
+
+import scala.util.Properties.envOrElse
+
+import org.apache.spark.SparkConf
+import org.apache.spark.streaming.Duration
+
+object CassandraConfig {
+  def init(conf: SparkConf, batchDuration: Duration): SparkConf = {
+    conf.setIfMissing("spark.cassandra.connection.host", envOrElse("FORTIS_CASSANDRA_HOST", ""))
+        .setIfMissing("spark.cassandra.auth.username", envOrElse("FORTIS_CASSANDRA_USER", ""))
+        .setIfMissing("spark.cassandra.auth.password", envOrElse("FORTIS_CASSANDRA_PASSWORD", ""))
+        .setIfMissing("spark.cassandra.connection.keep_alive_ms", (batchDuration.milliseconds * 2).toString)
+  }
+}

src/main/scala/com/microsoft/partnercatalyst/fortis/spark/sinks/cassandra/CassandraSchema.scala

@@ -0,0 +1,98 @@
+package com.microsoft.partnercatalyst.fortis.spark.sinks.cassandra
+
+import java.time.Instant.now
+import java.util.UUID
+
+import com.microsoft.partnercatalyst.fortis.spark.dto.AnalyzedItem
+import com.microsoft.partnercatalyst.fortis.spark.sinks.cassandra.Utils.{mean, rescale}
+import com.microsoft.partnercatalyst.fortis.spark.transforms.gender.GenderDetector.{Female, Male}
+import com.microsoft.partnercatalyst.fortis.spark.transforms.sentiment.SentimentDetector.Neutral
+
+case class Sentiment(
+  pos_avg: Float,
+  neg_avg: Float)
+
+case class Gender(
+  male_mentions: Int,
+  female_mentions: Int)
+
+case class Entities(
+  name: String,
+  externalsource: String,
+  externalrefid: String,
+  count: Float)
+
+case class Features(
+  mentions: Int,
+  sentiment: Sentiment,
+  gender: Gender,
+  entities: Set[Entities])
+
+case class Event(
+  pipeline: String,
+  externalid: String,
+  computedfeatures: Features,
+  detectedkeywords: Set[String],
+  detectedplaceids: Set[String],
+  event_time: Long,
+  eventlangcode: String,
+  id: UUID,
+  insertion_time: Long,
+  messagebody: String,
+  sourceid: String,
+  sourceurl: String,
+  title: String)
+
+object CassandraSchema {
+  def apply(item: AnalyzedItem): Event = {
+    Event(
+      pipeline = item.publisher,
+      externalid = "", // todo
+      computedfeatures = getFeature(item),
+      detectedkeywords = item.analysis.keywords.map(_.name).toSet,
+      detectedplaceids = item.analysis.locations.map(_.wofId).toSet,
+      event_time = item.createdAtEpoch,
+      eventlangcode = item.analysis.language.orNull,
+      id = item.id,
+      insertion_time = now.getEpochSecond,
+      messagebody = item.body,
+      sourceid = "", // todo
+      sourceurl = item.sourceUrl,
+      title = item.title)
+  }
+
+  private def getFeature(item: AnalyzedItem): Features = {
+    val genderCounts = item.analysis.genders.map(_.name).groupBy(identity).mapValues(_.size)
+    val entityCounts = item.analysis.entities.map(_.name).groupBy(identity).mapValues(_.size)
+    val positiveSentiments = item.analysis.sentiments.filter(_ > Neutral)
+    val negativeSentiments = item.analysis.sentiments.filter(_ < Neutral)
+    Features(
+      mentions = -1,
+      sentiment = Sentiment(
+        pos_avg = if (positiveSentiments.nonEmpty) mean(rescale(positiveSentiments, 0, 1)).toFloat else -1,
+        neg_avg = if (negativeSentiments.nonEmpty) mean(rescale(negativeSentiments, 0, 1)).toFloat else -1),
+      gender = Gender(
+        male_mentions = genderCounts.getOrElse(Male, -1),
+        female_mentions = genderCounts.getOrElse(Female, -1)),
+      entities = entityCounts.map(kv => Entities(
+        name = kv._1,
+        count = kv._2,
+        externalsource = "", // todo
+        externalrefid = "" // todo
+      )).toSet)
+  }
+}
+
+object Utils {
+  def mean(items: List[Double]): Double = {
+    items.sum / items.length
+  }
+
+  /** @see https://stats.stackexchange.com/a/25897 */
+  def rescale(items: List[Double], min_new: Double, max_new: Double): List[Double] = {
+    val min_old = items.min
+    val max_old = items.max
+    val coef = (max_new - min_new) / (max_old - min_old)
+    items.map(v => coef * (v - max_old) + max_new)
+  }
+}

src/main/scala/com/microsoft/partnercatalyst/fortis/spark/sinks/cassandra/CassandraSink.scala

@@ -0,0 +1,13 @@
+package com.microsoft.partnercatalyst.fortis.spark.sinks.cassandra
+
+import com.microsoft.partnercatalyst.fortis.spark.dto.AnalyzedItem
+import org.apache.spark.streaming.dstream.DStream
+import com.datastax.spark.connector.streaming._
+
+object CassandraSink {
+  def apply(dstream: Option[DStream[AnalyzedItem]], keyspaceName: String, tableName: String): Unit = {
+    if (dstream.isDefined) {
+      dstream.get.map(CassandraSchema(_)).saveToCassandra(keyspaceName, tableName)
+    }
+  }
+}

src/main/scala/com/microsoft/partnercatalyst/fortis/spark/transforms/gender/GenderDetector.scala

@@ -0,0 +1,6 @@
+package com.microsoft.partnercatalyst.fortis.spark.transforms.gender
+
+object GenderDetector {
+  val Male = "M"
+  val Female = "F"
+}

src/test/scala/com/microsoft/partnercatalyst/fortis/spark/IntegrationTestSpec.scala

@@ -1,11 +1,13 @@
 package com.microsoft.partnercatalyst.fortis.spark
 
+import scala.util.Properties.{envOrElse, envOrNone}
+
 import org.scalatest.FlatSpec
 
 class IntegrationTestSpec extends FlatSpec {
   protected def checkIfShouldRunWithLocalModels(): Option[String] = {
-    val runIntegrationTests = Option(System.getenv("FORTIS_INTEGRATION_TESTS")).getOrElse("false").toBoolean
-    val localModels = Option(System.getenv("FORTIS_MODELS_DIRECTORY"))
+    val runIntegrationTests = envOrElse("FORTIS_INTEGRATION_TESTS", "false").toBoolean
+    val localModels = envOrNone("FORTIS_MODELS_DIRECTORY")
     if (!runIntegrationTests && localModels.isEmpty) {
       cancel("Integration tests disabled and no local models available")
     }