Structured Streaming - Datasets

문제

Dataset 을 Streaming 에 활용해 보자

 

코드

   import spark.implicits._
   val carsDS: Dataset[Car] = spark.readStream
      .format("socket")
      .option("host", "localhost")
      .option("port", 12345)
      .load() // DF with single string column "value"
      .select(from_json(col("value"), carsSchema).as("car")) // composite column (struct)
      .selectExpr("car.*") // DF with multiple columns
      .as[Car] // encoder can be passed implicitly with spark.implicits
      
    val carCountByOrigin = carsDS.groupByKey(car => car.Origin).count()
    
    carCountByOrigin
      .writeStream
      .format("console")
      .outputMode("complete")
      .start()
      .awaitTermination()

 

참고: https://www.udemy.com/course/spark-streaming