structured streaming 窗口操作解析和测试

╰+攻爆jí腚メ 2023-02-24 12:29 15阅读 0赞

> 参考资料  
> 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 直白些说只会有更新的数据的结果表

[image.png]: /images/20230209/b1b5010f64fc45b3996123898d9580a1.png