目錄

???????Operations 操作

?

官網示例代碼：

---

Operations 操作

獲得到Source之后的基本數據處理方式和之前學習的DataFrame、DataSet一致，不再贅述

?

?

官網示例代碼：

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 10df.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 typedf.groupBy("deviceType").count()?????????????????// using untyped API// Running average signal for each device typeimport?org.apache.spark.sql.expressions.scalalang.typedds.groupByKey(_.deviceType).agg(typed.avg(_.signal))????// using typed API

?

總結

以上是生活随笔為你收集整理的2021年大数据Spark（四十六）：Structured Streaming Operations 操作的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。