参考资料
https://spark.apache.org/docs/2.4.1/structured-streaming-programming-guide.html
概念解析
structured streaming 窗口操作类似于分组操作，窗口操作都会维护每一个窗口所属的时间。
这边上一个官方的图
image.png
具体示例代码
我这边直接消费的kafka的数据，然后引用的kafka数据中的timestamp，然后将kafka的value简单的做了个聚合操作然后将后续结果输出到控制台，样例代码如下
val lines = spark.readStream.format("kafka")
.option("kafka.bootstrap.servers","10.130.7.208:9092")
.option("startingOffsets", "earliest")
.option("subscribe","test")
.load()
val value = lines.selectExpr("CAST(value AS STRING)","CAST(timestamp AS TIMESTAMP)").as[(String,String)]
import org.apache.spark.sql.functions._
val windowedCounts = value.groupBy(
window($"timestamp", "2 minute", "1 minute"),
$"value"
).count()
val query = windowedCounts
.writeStream
.outputMode("complete")
.format("console")
//"truncate"只是为了在控制台输出时，不进行列宽度自动缩小。
.option("truncate", "false")
.start()
注意点
这个样例代码注意这么几点(自己在上面碰到的坑和发现)
1、window窗口函数爆红不能使用
原因是没有导入 import org.apache.spark.sql.functions._
2、window窗口函数解析
window(timeColumn: Column, windowDuration: String, slideDuration: String)
注意几个点就是
timeColumn 的类型必须是 TIMESTAMP类型
windowDuration 窗口持续时间
slideDuration 窗口滑动时间 这个值要小于或者等于windowDuration
3、输出outputMode的格式选择
在窗口模式下的聚合等操作是，不可以使用append模式，不然报错。
在complete和update中的区别是
complete模式------ 每次有更新回将所有的行写入到sink，直白些说就是这种模式下会更新全部的结果表
update模式--------只会将更新的row写入到sink 直白些说只会有更新的数据的结果表
网友评论