Add lessons course

Author: Christian Herrera

.gitignore

@@ -0,0 +1,15 @@
+dist/*
+target/
+spark-warehouse/
+project/boot/
+project/plugins/project/
+.cache
+
+### Scala ###
+*.class
+*.log
+
+# Embedded metastore
+derby.log
+metastore_db/
+

.scalafmt.conf

@@ -0,0 +1,4 @@
+version=2.5.2
+align.preset = more
+maxColumn = 80
+importSelectors = singleLine

README.md

@@ -0,0 +1,23 @@
+<p align="center">
+  <a href="https://codely.com">
+    <img src="https://user-images.githubusercontent.com/10558907/170513882-a09eee57-7765-4ca4-b2dd-3c2e061fdad0.png" width="300px" height="92px" alt="Codely logo"/>
+  </a>
+</p>
+
+<h1 align="center">
+    🎇 Spark for programmers
+</h1>
+
+<p align="center">
+    <a href="https://github.com/CodelyTV"><img src="https://img.shields.io/badge/Codely-OS-green.svg?style=flat-square" alt="Codely Open Source projects"/></a>
+    <a href="https://pro.codely.com"><img src="https://img.shields.io/badge/Codely-Pro-black.svg?style=flat-square" alt="Codely Pro courses"/></a>
+</p>
+
+<p align="center">
+    Learn Apache Spark from scratch with a practical approach
+</p>
+
+<p align="center">
+  <a href="https://github.com/CodelyTV/spark_for_devs-course/stargazers">Stars are welcome 😊</a><br><br>
+  Course (Spanish): <a href="#">Spark for programmers</a>
+</p>

build.sbt

@@ -0,0 +1,38 @@
+scalaVersion := "2.12.12"
+version := "0.1.0-SNAPSHOT"
+name := "spark-for-programmers-course"
+organization := "com.codely"
+
+val sparkVesion = "3.5.0"
+
+libraryDependencies ++= Seq(
+  "org.apache.spark" %% "spark-core"           % sparkVesion,
+  "org.apache.spark" %% "spark-sql"            % sparkVesion,
+  "org.apache.spark" %% "spark-hive"           % sparkVesion,
+  "org.apache.spark" %% "spark-streaming"      % sparkVesion,
+  "org.apache.spark" %% "spark-sql-kafka-0-10" % sparkVesion,
+  "io.delta"         %% "delta-spark"          % "3.1.0",
+  // "com.amazonaws"     % "aws-java-sdk-bundle"        % "1.11.375",
+  "org.apache.hadoop" % "hadoop-aws"  % "3.2.2",
+  "com.rabbitmq"      % "amqp-client" % "5.12.0",
+  "com.typesafe"      % "config"      % "1.4.1",
+  //"org.apache.hadoop" % "hadoop-common"              % "3.3.1",
+  "org.scalatest" %% "scalatest"                       % "3.2.18"   % Test,
+  "org.scalatest" %% "scalatest-flatspec"              % "3.2.18"   % Test,
+  "com.dimafeng"  %% "testcontainers-scala"            % "0.40.12"  % Test,
+  "com.dimafeng"  %% "testcontainers-scala-kafka"      % "0.40.12"  % Test,
+  "com.dimafeng"  %% "testcontainers-scala-postgresql" % "0.41.4"   % Test,
+  "org.postgresql" % "postgresql"                      % "9.4.1207" % Test,
+  "org.mockito"   %% "mockito-scala"                   % "1.16.42"  % Test
+)
+
+assembly / mainClass := Some(
+  "com.codely.lesson_07_spark_optimize_and_monitoring.video_01__deploy_application.DeploySparkApp"
+)
+
+assembly / assemblyMergeStrategy := {
+  case PathList("META-INF", xs @ _*) => MergeStrategy.discard
+  case PathList("org", "apache", "spark", "unused", "UnusedStubClass.class") =>
+    MergeStrategy.first
+  case _ => MergeStrategy.first
+}

project/build.properties

@@ -0,0 +1 @@
+sbt.version = 1.9.8

project/plugins.sbt

@@ -0,0 +1,2 @@
+addSbtPlugin("com.eed3si9n"  % "sbt-assembly" % "1.2.0")
+addSbtPlugin("org.scalameta" % "sbt-scalafmt" % "2.5.2")

src/main/scala/com/codely/lesson_01__discover_apache_spark/video_01__from_excel_to_sql/FromCSVToSQL.scala

@@ -19,7 +19,8 @@ object FromCSVToSQL extends App {
     .csv(pathNetflixFile)
     .createOrReplaceTempView("netflix")
 
-  spark.sql("select * from netflix")
+  spark
+    .sql("select * from netflix")
     .show()
 
   // Making use of the csv data source options
@@ -32,7 +33,8 @@ object FromCSVToSQL extends App {
     .csv(pathNetflixFile)
     .createOrReplaceTempView("netflix")
 
-  spark.sql("SELECT * FROM netflix LIMIT 10")
+  spark
+    .sql("SELECT * FROM netflix LIMIT 10")
     .show()
 
   spark
@@ -60,7 +62,8 @@ object FromCSVToSQL extends App {
         |""".stripMargin)
     .show()
 
-  spark.sql("""
+  spark
+    .sql("""
       | SELECT lower(word), count(*) AS count
       |   FROM (
       |     SELECT explode(split(title, ' ')) as word from netflix

src/main/scala/com/codely/lesson_01__discover_apache_spark/video_02__trafffic_bottleneck_detection/TrafficBottleneckDetection.scala

@@ -0,0 +1,44 @@
+package com.codely.lesson_01__discover_apache_spark.video_02__trafffic_bottleneck_detection
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.functions.col
+import org.apache.spark.sql.streaming.Trigger
+
+private object TrafficBottleneckDetection extends App {
+
+  val SpeedThreshold = 60
+  val SocketPort     = 9999
+
+  val spark = SparkSession.builder
+    .master("local[*]")
+    .getOrCreate()
+
+  spark.sparkContext.setLogLevel("WARN")
+
+  val lines = spark.readStream
+    .format("socket")
+    .option("host", "localhost")
+    .option("port", SocketPort)
+    .load()
+
+  import spark.implicits._
+
+  val trafficData = lines
+    .as[String]
+    .map(line => line.split(","))
+    .map(arr => (arr(0), arr(1).toInt))
+    .toDF("segmentID", "speed")
+
+  val averageSpeeds = trafficData.groupBy("segmentID").avg("speed")
+
+  val trafficJams = averageSpeeds.filter(col("avg(speed)") < SpeedThreshold)
+
+  val query = trafficJams.writeStream
+    .format("console")
+    .trigger(Trigger.ProcessingTime("5 seconds"))
+    .outputMode("complete")
+    .start()
+
+  query.awaitTermination()
+
+}

src/main/scala/com/codely/lesson_01__discover_apache_spark/video_02__trafffic_bottleneck_detection/TrafficDataGenerator.scala

@@ -0,0 +1,47 @@
+package com.codely.lesson_01__discover_apache_spark.video_02__trafffic_bottleneck_detection
+
+import java.io.PrintStream
+import java.net.ServerSocket
+import scala.util.Random
+
+private object TrafficDataGenerator extends App {
+  val Port        = 9999
+  val MinSpeed    = 10
+  val MaxSpeed    = 110
+  val NumSegments = 5
+  val RefreshRate = 500 // Milliseconds
+
+  val server = new ServerSocket(Port)
+  println(s"Data generator server started at port $Port...")
+
+  val client = server.accept()
+  println("Client connected.")
+
+  val out = new PrintStream(client.getOutputStream)
+
+  try {
+    while (true) {
+      sendTrafficData(generateTrafficData)
+      Thread.sleep(RefreshRate)
+    }
+  } catch {
+    case e: Exception => e.printStackTrace()
+  } finally {
+    out.close()
+    client.close()
+    server.close()
+    println("Server stopped.")
+  }
+
+  private def sendTrafficData(data: String): Unit = {
+    println(s"Sending data: $data")
+    out.println(data)
+    out.flush()
+  }
+
+  private def generateTrafficData = {
+    val segmentId = Random.nextInt(NumSegments) + 1                    // 1 to 5
+    val speed     = Random.nextInt(MaxSpeed - MinSpeed + 1) + MinSpeed // 10 to 110
+    s"$segmentId,$speed"
+  }
+}

src/main/scala/com/codely/lesson_01__discover_apache_spark/video_03__intro_domain_events_analysis/HighPriceProductsPurchased.scala

@@ -0,0 +1,93 @@
+package com.codely.lesson_01__discover_apache_spark.video_03__intro_domain_events_analysis
+
+import org.apache.spark.sql.{DataFrame, SparkSession}
+import org.apache.spark.sql.functions.{col, desc, explode, lit, month}
+import org.apache.spark.sql.types._
+
+private object HighPriceProductsPurchased extends App {
+
+  val spark = SparkSession
+    .builder()
+    .appName("HighPriceProductsPurchased")
+    .master("local[8]")
+    .getOrCreate()
+
+  spark.sparkContext.setLogLevel("WARN")
+
+  val purchasedCompletedFilePath =
+    "src/main/scala/com/codely/lesson_01__discover_apache_spark/video_03__intro_domain_events_analysis/data/purchasecompleted.json"
+
+  spark.read
+    .format("json")
+    .load(purchasedCompletedFilePath)
+
+  val productPurchasedDF: DataFrame = spark.read
+    .json(purchasedCompletedFilePath)
+
+  productPurchasedDF.show()
+  productPurchasedDF.printSchema()
+
+  val productPurchasedSchema: StructType = StructType(
+    Array(
+      StructField("eventId", StringType),
+      StructField("eventType", StringType),
+      StructField(
+        "products",
+        ArrayType(
+          StructType(
+            Array(
+              StructField("productId", StringType),
+              StructField("quantity", IntegerType),
+              StructField("description", StringType),
+              StructField("category", StringType),
+              StructField("price", FloatType)
+            )
+          )
+        )
+      ),
+      StructField("timestamp", TimestampType),
+      StructField("transactionId", StringType),
+      StructField("userId", StringType)
+    )
+  )
+
+  val productPurchasedWithSchemaDF = spark.read
+    .schema(productPurchasedSchema)
+    .json(purchasedCompletedFilePath)
+
+  productPurchasedWithSchemaDF.printSchema()
+
+  productPurchasedWithSchemaDF
+    .select(col("transactionId"))
+    .show()
+
+  productPurchasedWithSchemaDF
+    .filter(month(col("timestamp")) === 2)
+    .show()
+
+  productPurchasedWithSchemaDF
+    .withColumn("new_column", lit("codely"))
+    .show(false)
+
+  val februaryTransactions =
+    productPurchasedWithSchemaDF.filter(month(col("timestamp")) === 2)
+
+  val explodedTransactions = februaryTransactions
+    .withColumn("product", explode(col("products")).as("product"))
+    .select(
+      col("timestamp"),
+      col("transactionId"),
+      col("product.description"),
+      col("product.category"),
+      col("product.price")
+    )
+
+  explodedTransactions.show(false)
+
+  explodedTransactions
+    .filter(col("category").isin("Electronics", "Gaming"))
+    .orderBy(desc("price"))
+    .dropDuplicates("description")
+    .limit(5)
+    .show(false)
+}