Added Kafka Producer

Author: novaonline

.gitignore

@@ -4,4 +4,4 @@
 
 /main/src/HueLogging.ServiceV2/Properties/launchSettings.json
 
-*.env
+.env

docker-compose.yml

@@ -9,7 +9,7 @@ services:
     ports:
       - "9092:9092"
     environment:
-      KAFKA_ADVERTISED_HOST_NAME: ${HueLogging:MachineHostIp}
+      KAFKA_ADVERTISED_HOST_NAME: ${ADVERTISED_HOST_NAME}
       KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
     volumes:
       - /var/run/docker.sock:/var/run/docker.sock
@@ -19,9 +19,9 @@ services:
       - "9042:9042"
       - "7000:7000"
     environment:
-      CASSANDRA_BROADCAST_ADDRESS: ${HueLogging:MachineHostIp}
+      CASSANDRA_BROADCAST_ADDRESS: ${ADVERTISED_HOST_NAME}
     volumes:
-      - ${HueLogging:CassandraVolumePath}:/var/lib/cassandra
+      - ${HueLogging_CassandraVolumePath}:/var/lib/cassandra
   polling-service:
     build:
       context: ./main/src

run.sh

@@ -0,0 +1,28 @@
+#!/bin/bash
+
+streamName="hue-logging-stream"
+streamVersion="1.0-SNAPSHOT"
+streamBasePath="stream/${streamName}"
+streamJarPath="${streamBasePath}/target/${streamName}-${streamVersion}.jar"
+
+# Before running this, please make sure that you have a .env file and the expected variables exist. See sample.env for expected variables
+# TODO fix the hostname issue. Services should be able to run and communicate using hostname/alias
+
+# First we need to build stream using Maven
+echo "Building Stream Job"
+mvn -f "${streamBasePath}/pom.xml" clean package 
+if [ $? -eq 0 ] 
+   # Now we need to spawn up all services
+   then
+    echo "Spawning up all services" 
+    docker-compose up -d 
+fi
+if [ $? -eq 0 ]
+   then 
+    # Then we start the stream job
+    echo "Starting Stream Job"
+    JOBMANAGER_CONTAINER=$(docker ps --filter name=jobmanager --format={{.ID}})
+    docker cp ${streamJarPath} "$JOBMANAGER_CONTAINER":/job.jar
+    CASSANDRA_HOST_IP=`docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' $(docker ps --filter name=cassandra --format={{.ID}})`
+    docker exec -it -d "$JOBMANAGER_CONTAINER" flink run /job.jar --cassandraHost `echo $CASSANDRA_HOST_IP`
+fi

sample.env

@@ -0,0 +1,3 @@
+HueLogging:ApiKey= #Hue Api Key that can be generated using CLI
+ADVERTISED_HOST_NAME= # Currently IP address has been tested and works. Still need to investigate why hostname does not
+HueLogging_CassandraVolumePath= # Filepath to the cassandra volume where data is stored

stop.sh

@@ -0,0 +1,2 @@
+#!/bin/bash
+docker-compose down

stream/hue-logging-stream/pom.xml

@@ -111,6 +111,13 @@ under the License.
             <version>1.2.17</version>
             <scope>runtime</scope>
         </dependency>
+
+        <!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.module/jackson-module-scala -->
+        <dependency>
+            <groupId>com.fasterxml.jackson.module</groupId>
+            <artifactId>jackson-module-scala_${scala.binary.version}</artifactId>
+            <version>2.9.8</version>
+        </dependency>
     </dependencies>
 
     <build>

stream/hue-logging-stream/src/main/scala/com/github/novaonline/LightEvents.scala

@@ -18,19 +18,19 @@
 
 package com.github.novaonline
 
+import java.time.OffsetDateTime
 import java.util.Properties
 
 import com.github.novaonline.cassandra.HueLoggingLocalCluster
-import com.github.novaonline.map.JsonToLightEvent
 import com.github.novaonline.model.light._
+import com.github.novaonline.serialization.SimpleJsonSchema
 import org.apache.flink.api.java.utils.ParameterTool
 import org.apache.flink.streaming.api.TimeCharacteristic
 import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
 import org.apache.flink.streaming.api.scala._
 import org.apache.flink.streaming.api.windowing.time.Time
 import org.apache.flink.streaming.connectors.cassandra.CassandraSink
-import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010
-import org.apache.flink.streaming.util.serialization.JSONKeyValueDeserializationSchema
+import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer010, FlinkKafkaProducer010}
 import org.slf4j.LoggerFactory
 
 
@@ -50,32 +50,39 @@ object LightEvents {
     val env = StreamExecutionEnvironment.getExecutionEnvironment
     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
 
+    val lightEventSchema = new SimpleJsonSchema[LightEvent]
+    val lightSessionSchema = new SimpleJsonSchema[LightSession]
+    val lightAccumlatedSchema = new SimpleJsonSchema[LightAccumulated]
+
     val properties = new Properties()
     properties.setProperty("bootstrap.servers", s"$kafkaHost:9092")
     properties.setProperty("group.id", "Hue-Logging")
-    val lightEventSource = new FlinkKafkaConsumer010("hue-logging-light-event", new JSONKeyValueDeserializationSchema(false), properties)
+    val lightEventSource = new FlinkKafkaConsumer010("hue-logging-light-event", lightEventSchema, properties)
     lightEventSource.setStartFromEarliest()
 
     val cassandraCluster = new HueLoggingLocalCluster(cassandraHost)
 
     // Start Map Reduce
-    val lightEventJsonStream = env.addSource(lightEventSource)
-    val lightEventStream = lightEventJsonStream.map(JsonToLightEvent)
+    val lightEventStream = env.addSource(lightEventSource).map(x => {
+      // enrich anything else missed from serializing
+      x.copy(state = x.state.copy(addDate = OffsetDateTime.parse(x.addDate).toEpochSecond))
+    })
 
     val watermarkedLightEventsStream = lightEventStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LightEvent](Time.seconds(10)) {
-      override def extractTimestamp(t: LightEvent): Long = t.state.addDate
+      override def extractTimestamp(t: LightEvent): Long = OffsetDateTime.parse(t.addDate).toEpochSecond
     })
 
     val lightSessionsStream = watermarkedLightEventsStream
       .keyBy(x => x.light.id)
-      .flatMapWithState[LightSession, LightState]((light: LightEvent, state: Option[LightState]) => {
-      if (state.isEmpty && light.state.on) {
+      .flatMapWithState[LightSession, LightState]((light: LightEvent, prevState: Option[LightState]) => {
+       val currState = light.state
+      if (prevState.isEmpty && currState.on) {
         LOG.info("Opened Light Session")
-        (Iterator(LightSession(light.light, light.state, None, 0)), Some(light.state))
-      } else if (state.isDefined && state.get.on && !light.state.on) {
+        (Iterator(LightSession(light.light, currState, None, 0)), Some(currState))
+      } else if (prevState.isDefined && prevState.get.on && !currState.on) {
         LOG.info("Closed Light Session")
-        val duration = light.state.addDate - state.get.addDate
-        (Iterator(LightSession(light.light, light.state, Some(light.state), duration)), None)
+        val duration = currState.addDate - prevState.get.addDate
+        (Iterator(LightSession(light.light, currState, Some(currState), duration)), None)
       } else {
         //LOG.info("Unknown State for Light Session")
         (Iterator.empty, None)
@@ -89,8 +96,17 @@ object LightEvents {
       .map(x => LightAccumulated(x.light, x.durationSeconds))
 
 
+    // Publish Processed Streams to Kafka
+    val kafkaLightSessionProducer = new FlinkKafkaProducer010("hue-logging-light-session", lightSessionSchema, properties)
+
+    val kafkaLightAccumulatedProducer = new FlinkKafkaProducer010("hue-logging-light-accumulated", lightAccumlatedSchema, properties)
+
+    lightSessionsStream.addSink(kafkaLightSessionProducer)
+    lightAccumulated.addSink(kafkaLightAccumulatedProducer)
+
+
     CassandraSink.addSink(lightEventStream.map(x => {
-      val r = x.toCassandraTuple()
+      val r = x.toCassandraTuple
       LOG.info(s"mapped light event stream ${r.toString()}")
       r
     }))

stream/hue-logging-stream/src/main/scala/com/github/novaonline/map/JsonToLightEvent.scala

@@ -1,12 +1,9 @@
 package com.github.novaonline.model.light
 
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode
-
 case class Light(
                   id: String,
                   hueType: String,
                   name: String,
                   modelId: String,
-                  sWVersion: String,
-                  raw: Option[JsonNode] = None
+                  sWVersion: String
                 )

stream/hue-logging-stream/src/main/scala/com/github/novaonline/model/light/Light.scala

@@ -1,6 +1,6 @@
 package com.github.novaonline.model.light
 
-case class LightEvent(light: Light, state: LightState)
+case class LightEvent(light: Light, state: LightState, addDate: String)
 {
-  def toCassandraTuple(): (String, String, Light, LightState ) = (light.name, light.id,  light, state )
+  def toCassandraTuple: (String, String, Light, LightState ) = (light.name, light.id,  light, state )
 }

stream/hue-logging-stream/src/main/scala/com/github/novaonline/model/light/LightEvent.scala

@@ -1,13 +1,10 @@
 package com.github.novaonline.model.light
 
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode
-
 case class LightState(
                        on: Boolean,
                        brightness: Int,
                        saturation: Int,
                        hue: Int,
                        reachable: Boolean,
-                       addDate: Long,
-                       raw: Option[JsonNode] = None
+                       addDate: Long
                      )

stream/hue-logging-stream/src/main/scala/com/github/novaonline/model/light/LightState.scala

@@ -0,0 +1,43 @@
+package com.github.novaonline.serialization
+
+import com.fasterxml.jackson.databind.{DeserializationFeature, MapperFeature, ObjectMapper}
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.typeutils.TypeExtractor
+import org.apache.flink.streaming.util.serialization.{KeyedDeserializationSchema, KeyedSerializationSchema}
+
+import scala.reflect.ClassTag
+
+class SimpleJsonSchema[T](implicit ct: ClassTag[T]) extends KeyedSerializationSchema[T] with KeyedDeserializationSchema[T] {
+
+  private var mapper: ObjectMapper = _
+
+  override def serializeKey(t: T): Array[Byte] = null
+
+  override def serializeValue(t: T): Array[Byte] = {
+    if (this.mapper == null) {
+      this.mapper = new ObjectMapper
+      mapper.registerModule(DefaultScalaModule)
+      mapper.configure(MapperFeature.ACCEPT_CASE_INSENSITIVE_PROPERTIES, true)
+      mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+
+    }
+    mapper.writeValueAsBytes(t)
+  }
+
+  override def getTargetTopic(t: T): String = null
+
+  override def deserialize(k: Array[Byte], v: Array[Byte], s: String, i: Int, l: Long): T = {
+    if (this.mapper == null) {
+      this.mapper = new ObjectMapper
+      mapper.registerModule(DefaultScalaModule)
+      mapper.configure(MapperFeature.ACCEPT_CASE_INSENSITIVE_PROPERTIES, true)
+      mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+    }
+    mapper.readValue(v, ct.runtimeClass.asInstanceOf[Class[T]])
+  }
+
+  override def isEndOfStream(t: T): Boolean = false
+
+  override def getProducedType: TypeInformation[T] = TypeExtractor.getForClass(ct.runtimeClass.asInstanceOf[Class[T]])
+}