2021年大数据Spark（四十六）：Structured Streaming Operations 操作

case class DeviceData(device: String, deviceType: String, signal: Double, time: DateTime)

val df: DataFrame = ... // streaming DataFrame with IOT device data with schema { device: string, deviceType: string, signal: double, time: string }

val ds: Dataset[DeviceData] = df.as[DeviceData]    // streaming Dataset with IOT device data

// Select the devices which have signal more than 10

df.select("device").where("signal > 10")      // using untyped APIs   

ds.filter(_.signal > 10).map(_.device)         // using typed APIs

// Running count of the number of updates for each device type

df.groupBy("deviceType").count()                 // using untyped API

// Running average signal for each device type

import org.apache.spark.sql.expressions.scalalang.typed

ds.groupByKey(_.deviceType).agg(typed.avg(_.signal))    // using typed API