前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >2,StructuredStreaming的事件时间和窗口操作

2,StructuredStreaming的事件时间和窗口操作

作者头像
Spark学习技巧
发布2018-06-22 11:33:30
1.1K0
发布2018-06-22 11:33:30
举报
文章被收录于专栏:Spark学习技巧

推荐阅读:1,StructuredStreaming简介

使用Structured Streaming基于事件时间的滑动窗口的聚合操作是很简单的,很像分组聚合。在一个分组聚合操作中,聚合值被唯一保存在用户指定的列中。在基于窗口的聚合的情况下,对于行的事件时间的每个窗口,维护聚合值。

如前面的例子,我们运行wordcount操作,希望以10min窗口计算,每五分钟滑动一次窗口。也即,12:00 - 12:10, 12:05 - 12:15, 12:10 - 12:20 这些十分钟窗口中进行单词统计。12:00 - 12:10意思是在12:00之后到达12:10之前到达的数据,比如一个单词在12:07收到。这个单词会影响12:00 - 12:10, 12:05 - 12:15两个窗口。

结果表将如下所示。

import org.apache.spark.sql.streaming.Trigger import java.sql.Timestamp import org.apache.spark.sql.functions._ import spark.implicits._ val lines = spark.readStream.format("socket").option("host", "127.0.0.1").option("port", 9999).option("includeTimestamp", true).load() val words = lines.as[(String, Timestamp)].flatMap(line =>line._1.split(" ").map(word => (word, line._2))).toDF("word", "timestamp") val windowedCounts = words.withWatermark("timestamp", "30 seconds").groupBy(window($"timestamp", "30 seconds", "15 seconds"), $"word").count() val query = windowedCounts.writeStream.outputMode("Append").format("console").trigger(Trigger.ProcessingTime(5000)).option("truncate", "false").start() query.awaitTermination()

本文参与 腾讯云自媒体同步曝光计划,分享自微信公众号。
原始发表:2018-05-11,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 浪尖聊大数据 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档