Flink时间窗口运用

上一篇介绍了Flink定时读取外部数据Flink 定时加载外部文件数据并广播

这一篇将介绍Flink定时输出到外部存储介质，有两种办法实现，一种是同上一篇一样，在RichXXXFunction中实现SinkFunction的方法，在其中open()方法中引入java的定时任务。

本文介绍另一种实现，基于Flink window窗口机制，将结果定时sink到外部文件。

需求：

经过flink清洗后的数据，要求每天sink一次数据到某文件中（该文件内容是json格式，需要进行追加，不属于flink范畴不细讲)。

实现：

1、数据清洗：

 DataStream<JSONObject> userSink = env.addSource(ssConsumer).map(new MapFunction<String, JSONObject>() {
//此处清洗数据，取实时数据中的某些字段public JSONObject map(String value) {JSONObject jsonObject = new JSONObject();JSONObject out = new JSONObject();try {jsonObject = JSON.parseObject(value);out.put(String.valueOf(jsonObject.get("subject_id")), jsonObject.get("subject_name"));} catch (Exception e) {JSONObject json = new JSONObject();jsonObject = json;logger.error("This value parse has error:", e);}return out;}}).filter(new FilterFunction<JSONObject>() {@Override
//此处过滤非Json格式数据public boolean filter(JSONObject value) throws Exception {if ("error".equals(value.getOrDefault("error", ""))) {return false;} else {return true;}}});

2、自定义sink，流数据输出到txt文件：

/*** @Author luran* @create 2020/4/13 18:36* @Desc*//*** 继承RichSinkFunction<String>类，其中List<JSONObject>为source端传到sink的数据类型，这个视Source端数据类型而定。*/
public class MyRishSinkFileWriter extends RichSinkFunction<List<JSONObject>> implements SinkFunction<List<JSONObject>> {private static String path;/*** open方法在sink第一次启动时调用，一般用于sink的初始化操作*/@Overridepublic void open(Configuration parameters) {try {super.open(parameters);path = PropertyReaderUtil.getStrValue("user.txt.path");File file = new File(path);if (!file.exists()) {file.createNewFile();}} catch (Exception e) {log.error("获取user.txt路径失败：", e);}}/*** invoke方法是sink数据处理逻辑的方法，source端传来的数据都在invoke方法中进行处理* 其中invoke方法中第一个参数类型与RichSinkFunctionList<JSONObject>中的泛型对应。第二个参数* 为一些上下文信息*/@Overridepublic void invoke(List<JSONObject> v, Context context) {try {JSONObject json = null;String s = JsonFileReaderUtil.readJsonData(path);if (s.length() == 0) {json = new JSONObject();} else {json = JSONObject.parseObject(s);}for (JSONObject d : v) {json.putAll(d);}String value = json.toJSONString();File file = new File(path);FileWriter fileWriter = new FileWriter(file);fileWriter.write(value);fileWriter.flush();fileWriter.close();} catch (IOException e) {log.error("写入user文件异常：", e);}}/*** close方法在sink结束时调用，一般用于资源的回收操作*/@Overridepublic void close() throws Exception {super.close();}
}

3、时间窗口的启用：

//由于用了时间窗口，输出肯定是List<>     DataStream<List<JSONObject>> dataBaseStream = userSink.windowAll(TumblingProcessingTimeWindows.of(Time.days(1))).process(new ProcessAllWindowFunction<JSONObject, List<JSONObject>, TimeWindow>() {@Overridepublic void process(Context context, Iterable<JSONObject> iterable, Collector<List<JSONObject>> collector) throws Exception {logger.info("进入时间窗口:"+LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));List<JSONObject> arrayList = new ArrayList<JSONObject>();iterable.forEach(single -> {arrayList.add(single);});if (arrayList.size() > 0) {collector.collect(arrayList);}}});dataBaseStream.print("userStream");dataBaseStream.addSink(new MyRishSinkFileWriter());

测试：

方便测试，先将时间改为每30秒执行，Time.seconds(30)，循环发送kafka数据：

第1个时间窗口到达：Iterable中集合了这30秒接收的所有实时数据，统一处理

第2个时间窗口达到：

总结：

Flink是实时处理，window机制可以认为是flink的批处理实现，因为需要等待水位线对齐触发timer。一般还会基于时间窗口做一些统计，如Flink按统计Kafka中每小时的数量并输出到MySQL 。

Flink实战—基于时间窗口定时输出sink相关推荐

Flink从入门到真香(12、Flink一大利器-时间窗口）
flink中支持多种窗口,包括:时间窗口,session窗口,统计窗口等等,能想到的基本都可以实现时间窗口(Time Windows) 最简单常用的窗口形式是基于时间的窗口,flink支持三种种时间 ...
使用Spark Streaming SQL基于时间窗口进行数据统计
1.背景介绍流式计算一个很常见的场景是基于事件时间进行处理,常用于检测.监控.根据时间进行统计等系统中.比如埋点日志中每条日志记录了埋点处操作的时间,或者业务系统中记录了用户操作时间,用于统计各种操 ...
【Flink系列2】时间窗口
引出对于流处理系统来说,流入的消息是无限的,所以对于聚合或是连接等操作,流处理系统需要对流入的消息进行分段,然后基于每一段数据进行聚合或是连接等操作.消息的分段即称为窗口,流处理系统支持的窗口有很多 ...
基于时间的访问控制列表
版权声明:原创作品,允许转载,转载时请务必以超链接形式标明文章原始出处 .作者信息和本声明.否则将追究法律责任.http://zpp2009.blog.51cto.com/730423/256204 ...
2021年大数据Flink（十九）：案例一基于时间的滚动和滑动窗口
目录案例一基于时间的滚动和滑动窗口需求代码实现案例一基于时间的滚动和滑动窗口需求 nc -lk 9999 有如下数据表示: 信号灯编号和通过该信号灯的车的数量 9,3 9,2 9,7 4 ...
flink定时读取mysql数据_flink时间系统系列之实例讲解：如何做定时输出
flink时间系统系列篇幅目录: 六.实例讲解:如何做定时输出今天为大家带来flink时间系统系列最后一篇实战篇,同样也是查漏补缺篇:如何做定时输出,首先说一下定时输出的需求背景,在flink流处理 ...
万字详述 Flink SQL 4 种时间窗口语义！（收藏）
DML:窗口聚合大家好我是老羊,由于窗口涉及到的知识内容比较多,所以博主先为大家说明介绍下面内容时的思路,大家跟着思路走.思路如下: ⭐ 先介绍 Flink SQL 支持的 4 种时间窗口 ⭐ 分别 ...
最新 Flink 1.13 时间和窗口（时间语义、Watermark、Window 窗口、Trigger）快速入门、详细教程
时间和窗口文章目录时间和窗口一.Flink 的三种时间语义二.水位线(Watermark) 1. Flink 中的 Watermark 机制 2. 如何生成水位线 3. 水位线的传递三.窗口 ...
第三课大数据技术之Fink1.13的实战学习-时间和窗口
第三课大数据技术之Fink1.13的实战学习-时间和窗口文章目录第三课大数据技术之Fink1.13的实战学习-时间和窗口第一节时间定义 1.1 Flink中的时间语义 1.2 两种时间语义 ...

Flink实战—基于时间窗口定时输出sink

Flink时间窗口运用

需求：

实现：

测试：

总结：

Flink实战—基于时间窗口定时输出sink相关推荐

最新文章

热门文章