spark-structured-streaming-01-files.scala

import org.apache.spark.sql.SparkSession

val spark = SparkSession.
  builder.
  master("local").
  appName("Spark Structured Streaming Demo").
  getOrCreate

spark.sparkContext.setLogLevel("ERROR")

val orders = spark.
  readStream.
  schema("order_id INT, order_date STRING, order_customer_id INT, order_status STRING").
  csv("/mnt/c/data/retail_db/orders")

val query = orders.
  writeStream.
  queryName("orders").
  format("memory").
  start

spark.sql("select * from orders").show
  

spark-structured-streaming-02-socket.scala

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

val spark = SparkSession.
  builder.
  master("local").
  appName("Get Department Traffic").
  getOrCreate

import spark.implicits._
spark.sparkContext.setLogLevel("ERROR")

val lines = spark.readStream.
  format("socket").
  option("host", "localhost").
  option("port", "9999").
  load

val departmentTraffic = lines.
  where(split(split($"value", " ")(6), "/")(1) === "department").
  select(split(split($"value", " ")(6), "/")(2).alias("department_name")).
  groupBy($"department_name").
  agg(count($"department_name").alias("department_count"))

val query = departmentTraffic.
  writeStream.
  queryName("department_count").
  outputMode("complete").
  format("memory").
  start

spark.sql("select * from department_count").show