大数据-玩转数据-Flink窗口

一、Flink 窗口 理解

在流处理应用中，数据是连续不断的，因此我们不可能等到所有数据都到了才开始处理。当然我们可以每来一个消息就处理一次，但是有时我们需要做一些聚合类的处理，例如：在过去的1分钟内有多少用户点击了我们的网页。在这种情况下，我们必须定义一个窗口，用来收集最近一分钟内的数据，并对这个窗口内的数据进行计算。

流式计算是一种被设计用于处理无限数据集的数据处理引擎，而无限数据集是指一种不断增长的本质上无限的数据集，而Window窗口是一种切割无限数据为有限块进行处理的手段。

在Flink中, 窗口(window)是处理无界流的核心. 窗口把流切割成有限大小的多个"存储桶"(bucket), 我们在这些桶上进行计算.

时间窗口
时间窗口包含一个开始时间戳(包括)和结束时间戳(不包括), 这两个时间戳一起限制了窗口的尺寸。在代码中, Flink使用TimeWindow这个类来表示基于时间的窗口. 这个类提供了key查询开始时间戳和结束时间戳的方法, 还提供了针对给定的窗口获取它允许的最大时间戳的方法(maxTimestamp())，时间窗口又分3种:滚动窗口、滑动窗口、会话窗口。

二、数据准备

准备一个WaterSensor类方便演示

package com.lyh.bean;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@NoArgsConstructor
@AllArgsConstructor
public class WaterSensor {
    private String id;
    private Long ts;
    private Integer vc;
}

三、时间滚动窗口

滚动窗口有固定的大小, 窗口与窗口之间不会重叠也没有缝隙.比如,如果指定一个长度为5分钟的滚动窗口, 当前窗口开始计算, 每5分钟启动一个新的窗口。滚动窗口能将数据流切分成不重叠的窗口，每一个事件只能属于一个窗口。

滚动窗口有固定的大小, 窗口与窗口之间不会重叠也没有缝隙.比如,如果指定一个长度为5分钟的滚动窗口, 当前窗口开始计算, 每5分钟启动一个新的窗口.
滚动窗口能将数据流切分成不重叠的窗口，每一个事件只能属于一个窗口
1.时间间隔可以通过: Time.milliseconds(x), Time.seconds(x), Time.minutes(x),等等来指定，2.我们传递给window函数的对象叫窗口分配器.

时间滚动窗口代码

package com.lyh.flink07;

import com.lyh.bean.WaterSensor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

public class Window_s {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.socketTextStream("hadoop100",9999)
                .map(line -> {
                    String[] data = line.split(",");
                    return new WaterSensor(
                            data[0],
                            Long.valueOf(data[1]),
                            Integer.valueOf(data[2])
                    );
                })
                .keyBy(WaterSensor::getId)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .process(new ProcessWindowFunction<WaterSensor, String,String, TimeWindow>() {
                    @Override
                    public void process(String key,
                                        Context ctx,
                                        Iterable<WaterSensor> elements,
                                        Collector<String> out) throws Exception {
                    List<WaterSensor> list  = toList(elements);
                        long starttime = ctx.window().getStart();
                        long endtime = ctx.window().getEnd();

                        out.collect("窗口：" + starttime + "  " + endtime + "  " + "key:" + key + "  " + "list:" + list);

                    }
                }).print();
        env.execute();
    }

    private static <T>List<T> toList(Iterable<T> it) {
        List<T>  list = new ArrayList<>();
        for (T t : it) {
            list.add(t);
            
        }
        return list;
    }
}

运行结果
在hadoop100 服务器
输入nc -lk 999 启动socket
在这里插入图片描述

消费结果：
在这里插入图片描述

四、时间滑动窗口

与滚动窗口一样, 滑动窗口也是有固定的长度. 另外一个参数我们叫滑动步长, 用来控制滑动窗口启动的频率.
所以, 如果滑动步长小于窗口长度, 滑动窗口会重叠. 这种情况下, 一个元素可能会被分配到多个窗口中
例如, 滑动窗口长度10分钟, 滑动步长5分钟, 则, 每5分钟会得到一个包含最近10分钟的数据。

时间滑动窗口代码

package com.lyh.flink07;

import com.lyh.bean.WaterSensor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

public class Window_s {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.socketTextStream("hadoop100",9999)
                .map(line -> {
                    String[] data = line.split(",");
                    return new WaterSensor(
                            data[0],
                            Long.valueOf(data[1]),
                            Integer.valueOf(data[2])
                    );
                })
                .keyBy(WaterSensor::getId)
//                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .window(SlidingProcessingTimeWindows.of(Time.seconds(5),Time.seconds(2)))
                .process(new ProcessWindowFunction<WaterSensor, String,String, TimeWindow>() {
                    @Override
                    public void process(String key,
                                        Context ctx,
                                        Iterable<WaterSensor> elements,
                                        Collector<String> out) throws Exception {
                    List<WaterSensor> list  = toList(elements);
                        long starttime = ctx.window().getStart();
                        long endtime = ctx.window().getEnd();

                        out.collect("窗口：" + starttime + "  " + endtime + "  " + "key:" + key + "  " + "list:" + list);

                    }
                }).print();
        env.execute();
    }

    private static <T>List<T> toList(Iterable<T> it) {
        List<T>  list = new ArrayList<>();
        for (T t : it) {
            list.add(t);
            
        }
        return list;
    }
}

执行结果
在hadoop100 服务器
输入nc -lk 999 启动socket
在这里插入图片描述
消费结果
在这里插入图片描述

五、时间会话窗口

会话窗口分配器会根据活动的元素进行分组. 会话窗口不会有重叠, 与滚动窗口和滑动窗口相比, 会话窗口也没有固定的开启和关闭时间.
如果会话窗口有一段时间没有收到数据, 会话窗口会自动关闭, 这段没有收到数据的时间就是会话窗口的gap(间隔)
我们可以配置静态的gap, 也可以通过一个gap extractor 函数来定义gap的长度. 当时间超过了这个gap, 当前的会话窗口就会关闭, 后序的元素会被分配到一个新的会话窗口。

时间会话窗口代码

package com.lyh.flink07;

import com.lyh.bean.WaterSensor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

public class Window_s {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.socketTextStream("hadoop100",9999)
                .map(line -> {
                    String[] data = line.split(",");
                    return new WaterSensor(
                            data[0],
                            Long.valueOf(data[1]),
                            Integer.valueOf(data[2])
                    );
                })
                .keyBy(WaterSensor::getId)
//                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
//                .window(SlidingProcessingTimeWindows.of(Time.seconds(5),Time.seconds(2)))
                .window(ProcessingTimeSessionWindows.withGap(Time.seconds(3)))
                .process(new ProcessWindowFunction<WaterSensor, String,String, TimeWindow>() {
                    @Override
                    public void process(String key,
                                        Context ctx,
                                        Iterable<WaterSensor> elements,
                                        Collector<String> out) throws Exception {
                    List<WaterSensor> list  = toList(elements);
                        long starttime = ctx.window().getStart();
                        long endtime = ctx.window().getEnd();

                        out.collect("窗口：" + starttime + "  " + endtime + "  " + "key:" + key + "  " + "list:" + list);

                    }
                }).print();
        env.execute();
    }

    private static <T>List<T> toList(Iterable<T> it) {
        List<T>  list = new ArrayList<>();
        for (T t : it) {
            list.add(t);
            
        }
        return list;
    }
}

运行结果

在hadoop100 服务器
输入nc -lk 999 启动socket
在这里插入图片描述
消费结果
在这里插入图片描述
因为会话窗口没有固定的开启和关闭时间, 所以会话窗口的创建和关闭与滚动,滑动窗口不同. 在Flink内部, 每到达一个新的元素都会创建一个新的会话窗口, 如果这些窗口彼此相距比较定义的gap小, 则会对他们进行合并. 为了能够合并, 会话窗口算子需要合并触发器和合并窗口函数: ReduceFunction, AggregateFunction, or ProcessWindowFunction 。

六、基于元素个数的滚动窗口

默认的CountWindow是一个滚动窗口，只需要指定窗口大小即可，当元素数量达到窗口大小时，就会触发窗口的执行。
实例代码
.countWindow(3)
说明:哪个窗口先达到3个元素, 哪个窗口就关闭. 不影响其他的窗口.

基于元素个数的滚动窗口代码

package com.lyh.flink07;

import com.lyh.bean.WaterSensor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

public class Window_s_n {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.socketTextStream("hadoop100",9999)
                .map(line -> {
                    String[] data = line.split(",");
                    return new WaterSensor(
                            data[0],
                            Long.valueOf(data[1]),
                            Integer.valueOf(data[2])
                    );
                })
                .keyBy(WaterSensor::getId)
                .countWindow(2)
                .process(new ProcessWindowFunction<WaterSensor, String, String, GlobalWindow>() {
                    @Override
                    public void process(String key,
                                        Context ctx,
                                        Iterable<WaterSensor> elements,
                                        Collector<String> out) throws Exception {
                        List<WaterSensor> list  = toList(elements);
                        out.collect("窗口：" + "key:" + key + "  " + "list:" + list);
                    }
                }).print();

        env.execute();
    }

    private static <T>List<T> toList(Iterable<T> it) {
        List<T>  list = new ArrayList<>();
        for (T t : it) {
            list.add(t);

        }
        return list;
    }
}

运行结果
在这里插入图片描述
在这里插入图片描述

七、基于元素个数的滑动窗口

滑动窗口和滚动窗口的函数名是完全一致的，只是在传参数时需要传入两个参数，一个是window_size，一个是sliding_size。下面代码中的sliding_size设置为了2，也就是说，每收到两个相同key的数据就计算一次，每一次计算的window范围最多是3个元素。
实例代码
.countWindow(3, 2)

package com.lyh.flink07;

import com.lyh.bean.WaterSensor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

public class Window_s_n {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.socketTextStream("hadoop100",9999)
                .map(line -> {
                    String[] data = line.split(",");
                    return new WaterSensor(
                            data[0],
                            Long.valueOf(data[1]),
                            Integer.valueOf(data[2])
                    );
                })
                .keyBy(WaterSensor::getId)
//                .countWindow(2)
                .countWindow(3,2)
                .process(new ProcessWindowFunction<WaterSensor, String, String, GlobalWindow>() {
                    @Override
                    public void process(String key,
                                        Context ctx,
                                        Iterable<WaterSensor> elements,
                                        Collector<String> out) throws Exception {
                        List<WaterSensor> list  = toList(elements);
                        out.collect("窗口：" + "key:" + key + "  " + "list:" + list);
                    }
                }).print();

        env.execute();
    }

    private static <T>List<T> toList(Iterable<T> it) {
        List<T>  list = new ArrayList<>();
        for (T t : it) {
            list.add(t);

        }
        return list;
    }
}

运行结果
在这里插入图片描述
在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：/a/93280.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

Java【手撕双指针】LeetCode 57. “两数之和“, 图文详解思路分析 + 代码

Java【手撕双指针】LeetCode 57. “两数之和“, 图文详解思路分析 + 代码

文章目录 前言一、两数之和1, 题目2, 思路分析3, 代码展示 前言 各位读者好, 我是小陈, 这是我的个人主页, 希望我的专栏能够帮助到你: &#x1f4d5; JavaSE基础: 基础语法, 类和对象, 封装继承多态, 接口, 综合小练习图书管理系统等 &#x1f4d7; Java数据结构: 顺序表, 链表…
阅读更多...
用好「留存」，闭环小程序运营链路

用好「留存」，闭环小程序运营链路

如何通过线上小程序获取用户线索&#xff0c;提高企业抗风险能力&#xff0c;建立有效的营销数字化系统一直是困扰每一个小程序开发者与运营者的问题。 当我们选择使用小程序设计自己的运营流程时&#xff0c;从「推广」到「转化」&#xff0c;再到最终的「留存」都是运营过程…
阅读更多...
C语言(第三十三天)

C语言(第三十三天)

3.1.2 画图推演 3.2 举例2&#xff1a;顺序打印一个整数的每一位 输入一个整数m&#xff0c;打印这个按照顺序打印整数的每一位。 比如&#xff1a; 输入&#xff1a;1234 输出&#xff1a;1 2 3 4 输入&#xff1a;520 输出&#xff1a;5 2 0 3.2.1 分析和代码实现 这个题目&a…
阅读更多...
无涯教程-分类算法 - Python实现函数

无涯教程-分类算法 - Python实现函数

为了在Python中实现SVM&#xff0c;无涯教程将从标准库导入开始&#xff0c;如下所示- import numpy as np import matplotlib.pyplot as plt from scipy import stats import seaborn as sns; sns.set() 接下来&#xff0c;从sklearn.dataset.sample_generator创建具有线性可…
阅读更多...
WordPress使用子主题插件 Child Theme Wizard，即使主题升级也能够保留以前主题样式

WordPress使用子主题插件 Child Theme Wizard，即使主题升级也能够保留以前主题样式

修改WordPress网站样式&#xff0c;主题升级会导致自己定义设置的网站样式丢失&#xff0c;还需要重新设置&#xff0c;很繁琐工作量大&#xff0c;发现在WordPress 中有Child Theme Wizard子主题插件&#xff0c;使用Child Theme Wizard子主题插件&#xff0c;即使主题升级&am…
阅读更多...
设计模式-桥接模式

设计模式-桥接模式

核心思想 适配器模式类似&#xff0c;以后也会遇到意思接近一样的设计模式。在开发中一般多个模式混用&#xff0c;且根据不同的场景进行搭配&#xff0c;桥接模式也是结构型模式将抽象的部分和实现的部分分离&#xff0c;使它们都可以独立的变化。通俗来说&#xff0c;就是通…
阅读更多...
初识Java 1-1 面向对象的语言

初识Java 1-1 面向对象的语言

目录 引用的作用 数据的储存 常见的数据储存方式 特殊储存的基本类型 数组 销毁对象 基本类型的作用域 对象的作用域 创建新类型 - class关键字 方法、参数和返回值 参数列表 编写程序 名称可见性 使用组件 static关键字 Java程序 编程风格&#xff08;驼峰式…
阅读更多...
本地化部署ChatGLM2-6B模型

本地化部署ChatGLM2-6B模型

本地化部署ChatGLM2-6B模型 简介硬件需求 环境部署安装Miniconda创建虚拟环境下载模型和源码安装依赖GPU部署CPU部署 运行程序GPU模式CPU模式命令行运行网页版运行API运行 简介 ChatGLM是清华大学开源的方案&#xff0c;中文效果还是很不错的。基于 General Language Model (G…
阅读更多...
聚类分析 | MATLAB实现基于AHC聚类算法可视化

聚类分析 | MATLAB实现基于AHC聚类算法可视化

聚类分析 | MATLAB实现基于AHC聚类算法可视化 目录 聚类分析 | MATLAB实现基于AHC聚类算法可视化效果一览基本介绍程序设计参考资料 效果一览 基本介绍 AHC聚类算法&#xff0c;聚类结果可视化&#xff0c;MATLAB程序。 Agglomerative Hierarchical Clustering&#xff08;自底…
阅读更多...
CG MAGIC分享如何3d Max新版本如何能在旧版本中打开呢？

CG MAGIC分享如何3d Max新版本如何能在旧版本中打开呢？

三维行业来说&#xff0c;无论是三维软件还是插件&#xff0c;都是在持续更新功能的。 3d Max这款软件&#xff0c;自然也不例外&#xff0c;不断推出新版本以提供更多强大的功能和工具。 随着新版本的发布&#xff0c;旧版本用户可能面临一个问题&#xff1a; 3d Max新版本…
阅读更多...
java练习8.100m小球落地

java练习8.100m小球落地

题目: 如一个小球从100米高度自由落下&#xff0c;每次落地后就反跳回原高度的一半。 那么求它在第10次落地时&#xff0c;共经过多少米&#xff1f;第10次反弹多高&#xff1f; public static void main(String[] args) {/*假如一个小球从100米高度自由落下&#xff0c;每次落…
阅读更多...
PHP自己的框架cookie()使用（完善篇七）

PHP自己的框架cookie()使用（完善篇七）

1、PHP自己的框架cookie() 2、cookie类&#xff08;CookieBase.php&#xff09; <?php class CookieBase {/*** 设置cookie*/public static function set($name, $value, $expire 3600, $path , $domain , $secure false, $httponly false) {setcookie($name, $valu…
阅读更多...
【InsCode】InsCode打造的JavaSE与Linux命令互融的伪Linux文件系统小项目

【InsCode】InsCode打造的JavaSE与Linux命令互融的伪Linux文件系统小项目

&#x1f9d1;‍&#x1f4bb;作者名称&#xff1a;DaenCode &#x1f3a4;作者简介&#xff1a;啥技术都喜欢捣鼓捣鼓&#xff0c;喜欢分享技术、经验、生活。 &#x1f60e;人生感悟&#xff1a;尝尽人生百味&#xff0c;方知世间冷暖。 &#x1f4d6;所属专栏&#xff1a;Ja…
阅读更多...
python网络爬虫指南二：多线程网络爬虫、动态内容爬取（待续）

python网络爬虫指南二：多线程网络爬虫、动态内容爬取（待续）

文章目录 一、多线程网络爬虫1.1 线程的基础内容、GIL1.2 创建线程的两种方式1.3 threading.Thread类1.4 线程常用方法和锁机制1.5 生产者-消费者模式1.5.1 生产者-消费者模式简介1.5.2 Condition 类协调线程 1.6 线程中的安全队列1.6 多线程爬取王者荣耀壁纸1.6.1 网页分析1.6…
阅读更多...
dolphinschedule配置企微告警服务（WeChat群组）

dolphinschedule配置企微告警服务（WeChat群组）

一、前置说明 ds配置好工作流后&#xff0c;比较重要的一个就是上线后的监控报警服务&#xff0c;如果你是基于企微作为协同办公的&#xff0c;WeChat群组预警必须是要安排上的&#xff0c;文章基于自建应用配合群组方式构建预警群&#xff0c;接入后&#xff0c;任务成功或者…
阅读更多...
Java注解与反射

Java注解与反射

&#x1f497;wei_shuo的个人主页 &#x1f4ab;wei_shuo的学习社区 &#x1f310;Hello World &#xff01; Java注解与反射 Java注解和反射是Java语言中两个强大的特性&#xff0c;它们可以一起使用以实现动态的、灵活的编程和元数据处理 注解 Java注解&#xff08;Annotatio…
阅读更多...
4.1011

4.1011

目录 四次挥手中收到乱序的FIN包会如何处理&#xff1f; 在 TIME_WAIT 状态的 TCP 连接&#xff0c;收到 SYN 后会发生什么&#xff1f; 四次挥手中收到乱序的FIN包会如何处理&#xff1f; 如果FIN报文比数据包先道道客户端&#xff0c;此时FIN是一个乱序报文&#xff0c;此时…
阅读更多...
【LeetCode-中等题】2. 两数相加

【LeetCode-中等题】2. 两数相加

文章目录 题目方法一&#xff1a;借助一个进制位&#xff0c;以及更新尾结点方法一改进&#xff1a;相比较第一种&#xff0c;给head一个临时头节点&#xff08;开始节点&#xff09;&#xff0c;最后返回的时候返回head.next&#xff0c;这样可以省去第一次的判断 题目 方法一…
阅读更多...
IdentityServer密码长度超长会导致跳转到登录页

IdentityServer密码长度超长会导致跳转到登录页

应用系统项目的安全要求越来越高&#xff0c;基本都是采取https等加密证书传输&#xff0c;无法使用https的&#xff0c;也是要求不能明文传输内容&#xff0c;因此做一些等保要求&#xff0c;密码需要加密后才能传输给服务端&#xff0c;所以前端会采取一些密码手段&#xff0…
阅读更多...
STM32移植ST77891.69寸屏幕并移植lvgl8.0.2（按键输入设备）一些心得

STM32移植ST77891.69寸屏幕并移植lvgl8.0.2（按键输入设备）一些心得

学习目标： 将ST7789(1.69寸圆角屏SPI)驱动移植+lvgl移植+按键当作输入设备 学习内容： 驱动移植lvgl移植按键移植软件使用正片开始： 先说说这块屏幕的介绍呗 ST7789屏幕是一种高性能的液晶显示屏，它具有高清晰度、高亮度、低功耗等优点。它采用了SPI接口通信，可以实现快速…
阅读更多...
最新文章

Copyright @ 2022~2023