Add practical exercise lesson 3

Author: Christian Herrera

project/Dependencies.scala

@@ -13,7 +13,10 @@ object Dependencies {
     "com.rabbitmq"            % "amqp-client"          % "5.21.0",
     "com.typesafe"            % "config"               % "1.4.1",
     "io.delta"               %% "delta-spark"          % "3.1.0",
-    "io.spray"               %% "spray-json"           % "1.3.6"
+    "io.spray"               %% "spray-json"           % "1.3.6",
+    "io.circe"               %% "circe-core"           % "0.14.7",
+    "io.circe"               %% "circe-generic"        % "0.14.9",
+    "io.circe"               %% "circe-parser"         % "0.14.9"
   )
   private val test = Seq(
     "org.scalatest" %% "scalatest" % "3.2.19",

src/main/com/codely/lesson_03_spark_streaming_sqs/z_practical_exercise/3-publish_message.sh

@@ -0,0 +1,19 @@
+aws events put-events \
+    --endpoint-url http://localhost:4566 \
+    --region us-east-1 \
+    --entries '[{
+        "EventBusName": "codely.domain_events",
+        "Source": "codely",
+        "DetailType": "user.registered",
+        "Detail": "{ \"detail_type\": \"userRegistered\", \"user_id\": \"123\", \"email\": \"[email protected]\", \"timestamp\": \"2023-07-21T10:00:00Z\" }"
+    }, {
+        "EventBusName": "codely.domain_events",
+        "Source": "codely",
+        "DetailType": "user.registered",
+        "Detail": "{ \"detail_type\": \"userRegistered\", \"user_id\": \"124\", \"email\": \"[email protected]\", \"timestamp\": \"2023-07-21T10:01:00Z\" }"
+    }, {
+        "EventBusName": "codely.domain_events",
+        "Source": "codely",
+        "DetailType": "user.registered",
+        "Detail": "{ \"detail_type\": \"other\", \"user_id\": \"125\", \"email\": \"[email protected]\", \"timestamp\": \"2023-07-21T10:02:00Z\" }"
+    }]'

src/main/com/codely/lesson_03_spark_streaming_sqs/z_practical_exercise/SQSReceiverSparkApp.scala

@@ -0,0 +1,44 @@
+package com.codely.lesson_03_spark_streaming_sqs.z_practical_exercise
+
+import io.circe.generic.auto._
+import io.circe.parser._
+import org.apache.spark.SparkConf
+import org.apache.spark.streaming.{Seconds, StreamingContext}
+import org.apache.spark.streaming.dstream.DStream
+
+object SQSReceiverSparkApp extends App {
+  private val sqsEndpoint = "http://localhost:4566"
+  private val region      = "us-east-1"
+  private val queueUrl    = "http://localhost:4566/000000000000/send_welcome_email_on_user_registered"
+
+  val conf = new SparkConf().setAppName("SQSReceiverSparkApp").setMaster("local[*]")
+
+  val ssc = new StreamingContext(conf, Seconds(5))
+
+  val receiver = new SQSSparkReceiver(sqsEndpoint, region, queueUrl)
+
+  val messages: DStream[String] = ssc.receiverStream(receiver)
+
+  case class Detail(detail_type: String, user_id: String, email: String, timestamp: String)
+  case class Event(detail: Detail)
+
+  val filteredMessages = messages.flatMap { message =>
+    decode[Event](message) match {
+      case Right(event) if event.detail.detail_type == "userRegistered" => Some(event)
+      case _                                                            => None
+    }
+  }
+
+  val eventsDStream = filteredMessages.map { event =>
+    (event.detail.user_id, event.detail.timestamp)
+  }
+
+  val windowedCounts = eventsDStream
+    .map { case (userId, _) => (userId, 1) }
+    .reduceByKeyAndWindow(_ + _, Seconds(300))
+
+  windowedCounts.print()
+
+  ssc.start()
+  ssc.awaitTermination()
+}

src/main/com/codely/lesson_03_spark_streaming_sqs/z_practical_exercise/SQSSparkReceiver.scala

@@ -0,0 +1,55 @@
+package com.codely.lesson_03_spark_streaming_sqs.z_practical_exercise
+
+import com.amazonaws.client.builder.AwsClientBuilder
+import com.amazonaws.services.sqs.model.{DeleteMessageRequest, ReceiveMessageRequest}
+import com.amazonaws.services.sqs.{AmazonSQS, AmazonSQSClientBuilder}
+import org.apache.spark.storage.StorageLevel
+import org.apache.spark.streaming.receiver.Receiver
+
+import scala.collection.JavaConverters._
+
+class SQSSparkReceiver(endpoint: String, region: String, queueUrl: String)
+    extends Receiver[String](StorageLevel.MEMORY_AND_DISK_2) {
+
+  private var sqsClient: AmazonSQS = _
+
+  override def onStart(): Unit = {
+
+    sqsClient = AmazonSQSClientBuilder
+      .standard()
+      .withEndpointConfiguration(
+        new AwsClientBuilder.EndpointConfiguration(
+          endpoint,
+          region
+        )
+      )
+      .build()
+
+    new Thread("SQS Receiver") {
+      override def run() {
+        receive()
+      }
+    }.start()
+  }
+
+  override def onStop(): Unit = {
+    // Any necessary cleanup
+  }
+
+  private def receive(): Unit = {
+    while (!isStopped()) {
+      val request = new ReceiveMessageRequest(queueUrl)
+        .withMaxNumberOfMessages(10)
+        .withWaitTimeSeconds(20)
+
+      val messages = sqsClient.receiveMessage(request).getMessages.asScala
+
+      for (message <- messages) {
+        store(message.getBody)
+        val deleteRequest =
+          new DeleteMessageRequest(queueUrl, message.getReceiptHandle)
+        sqsClient.deleteMessage(deleteRequest)
+      }
+    }
+  }
+}