038.尚硅谷_Flink-流处理API_Window API_窗口函数及其它可选API

现在已经知道就是怎么样去定义一个窗口分配器了啊，所以我们总结一下的话，所谓的这个窗口分配器啊，Window a sign，它就是点window方法里边的参数啊这个类型，那么本身弗link给我们提供了几种通用的window sign这几种啊，滚动、滑动、绘画，还有全局啊，大家知道这个全局窗口呢，就是相当于没有定义窗口结束的时间，或者说结束的条件必须去自定义啊，那这种global window后面我们说到就是count window的底层就是用global window来实现啊，那接下来啊，大家就知道了，在每一种类型的窗口创建的时候，一般怎么用呢？哎，那其实就是能用简写就用简写，所以这里边滚动时间窗口和滑动时间窗口我们都直接用这个点tap window直接创建就可以了，但是这个会画窗口呢，那没办法，没有简写方式，那你就只好调底层的点window啊，然后如果是这个看不同的时间语义对吧，定义不同的时间语义，然后session Windows with get那。

后边的这个技术窗口呢，我们就不要再去写底层的那个格global window了，就直接count window就OK啊，这个调用还是比较简单的，这里需要说明的一点就是我们定义这个时间窗口的时候，假如说你想用到offset，用到那个偏移量的话，那就没有办法在这里边直接用简写的方式定义了，对吧？因为你滚动窗口你想加一个这个偏移量作为参数的话，它这里边并不知道你你传两个参数是要给偏移量啊，我默认认为你是要给一个滑动窗口的啊，所以这里边就是这里边没有偏移量这一说，你如果想要指定，那就只能用底层的那种方式来指定，呃，那大家就就会发现，这是不是我们这个window操作就都已经做完了呢？我们已经知道怎么样去创建窗口了嘛，这其实没完，大家一定要注意啊，在真正的所谓的窗口操作，大家如果要是最后把这个啊代码直接提交运行啊，你看到它那个执行图的时候，点window这一步。

操，或者time window count window这一步操作啊，它并不是不是一部真正意义上的，就是我们说的这个可执行的这一个任务，真正可执行的任务，大家想想应该是什么呢？这是不是来了这个定义窗口，你想这是对数据的处理吗？其实不是，它只是类似于就像我们之前那个分组KY一样，对吧，把数据做一个分组分发，我们这里边的这个window呢，它不做分组，不做这个重分区，但是它是相当于做了一个什么，做了一个分桶，对吧，那就是数据你到底属于哪个桶里边做一个分法。那至于说你这个窗口到点的时候，我到底要去做什么计算，你到底是要算一个最大值最小值，还是要做算一个求和，还是算一个平均数，我现在根本不知道对吧？哎，所以大家注意啊，完整的一个窗口操作，大家回顾一下前面我们这个示例代码，完整的一个窗口操作是什么呢？就是啊，首先你可以这个K，也可以不K对吧？不K的话是WINDOW2那种操作嘛，啊就是首先一般情况我们是K啊，分组之后，然后呢，要有一步窗口分配器去指定它到底数据分到哪个窗口里边去，然后呢，再来一步具体的计算操作啊，就是比方说reduce对吧，做计算就像我们之前那个K之后K外，这不是这不是一个具体的计算，只是表示按照哈西扣的重分区分组而已，而这里边你如果要是说开窗之后呢，同样也还没有定义计算方法，所以必须要有后边的这样一个。

计算的过程，这两个合在一起，这才叫一个完整的窗口操作，窗口任务对吧？啊，所以呃，Flink底层它会有一个实现，叫做window operator，我就简写了啊，Window operator叫做一个算子对吧？呃，运算符算子，那么这个window operator它其实就包含这两部分，就是我们前面的窗口分配和后边的这个计算啊，那后边的这个计算其实是有一个专门的名称的，它叫做窗口函数啊，就是说你要定义窗口，然后还要定义窗口里边到底怎么操作啊，这个操作是用一个窗口函数来定义的，当然这里边的窗口函数是一个广义的，呃，窗口函数啊，就是它代表的就是到底你要做什么计算啊，那这里面呢，可以分成两大类，两大类一类是所谓的增量聚合函数，另外一类是全窗口函数。这里要怎么去。

理解呢，这个其实整体来讲就是看你计算的过程当中到底用哪种方式来做计算，首先我们看这个增量聚合函数，增量聚合函数顾名思义啊，它是增量的去做聚合，意思就是说我当前比方说这个九点到十点开了一个窗口，对吧？那大家会想到我当前这个窗口是不是要等到十点钟才会关，对吧，才会输出一个统计的计算结果，哎，那这时候就就有问题了，你像之前我们这个在批处理的时候啊，Spark里边那这个简单，因为我们这都是一批一批的数据都现成嘛，都放在那儿了，对吧？那那所以接下来其实我就是只要啊，呃，你这个数据来了之后啊，那那我这个到点我就把之前的这个所有数据都算，算一遍就完事了，哎，可是现在这个flink啊link，我现在这个数据并不是一批一批的，而是不停的来的，对吧，我现在九点到十点，我们不是说它是一个。

懂吗？我把这个桶创建好之后。那接下来这个数据就不停的往里边塞，那应该是九点哎零一啊，来的数据我放进去，9.05来的数据放进去，在这一个小时之内，这数据是不停来，来里边我要不停的把它放在这桶里边的，对吧？那这就有一个问题，就是说我是一直把它都放在桶里，等到十点的时候再去把它挨个都拿出来，去做一次计算，然后做做输出呢，还是直接就来一个数据就算一次，来一个数据就算一次呢？

这就涉及到了不同的这种操作方式了，好，那如果说你是所有的数据都等都都先放在这儿对吧？啊都都先存起来，等到这个到点要去输出的时候，我再做一次计算，这个就叫什么呢？这这种计算的方式就叫全窗口函数，而与之对应的，如果说我是来一个就计算一个，来一个就计算一个，这种方式就叫做增量聚合函数。啊，所以大家看这就相当于有点相当于什么呢？我们说当前这个窗口的处理是对无界的流做了一个有界流的截取，对吧？啊，截取了有界的一部分，那这个窗口函数的选择这两类就相当于是什么呢？有界流你到底是把它当成一批先攒齐了，然后做一个批处理，哎，这个就是窗口函数，全窗口函数对吧？啊，那另外还有一种方式，就是我可以在里边当前的这一个有界流里还是零处理，就是来一个就聚合一次，来一个就聚合一次，那那这个过程它有什么区别呢？比方说以这个求和为例啊，以sum为例，诶大家会想到你假如12345很多数都来了，哎，那增量聚合函数的这个过程是不是一来了之后，哎，我就叠加，现在只有一个一，那状态是一对吧，我要保持一个状态，这个状态就是之前所有数的和。

然后我只要有了这个状态，是不是之前的所有数据我根本不用存啊啊对吧，不用存这么多数据，全把它保存在这个桶里边，我只要有这个状态就够了，然后二来了之后呢，我不用等到最后这个窗口关闭的时候，我直接就在之前一基础上一加三，对吧，加到一个三啊，然后再来一个新的，我直接在之前巨头基础上直接加，然后等到这个窗口关闭的时候，要关闭的时候，你说我干什么事呢？我是说是不是直接拿出来当前已经加好的那个结果输出就完事了。

啊，所以大家看为什么流处理比这个批处理要快呢？就是因为在我们这里边统计的过程当中，在数据来的过程当中，就是攒批的，平常我们批处理攒一批，等这个数据收集齐的过程当中，我已经在做叠加了，我已经把它算完了，对吧？我只是最后等到这个到点的时候把结果输出一下而已，而批处理呢，那就是要等到所有数据都到齐，然后每来一个数，就是等到这个，呃，最后要输出结果的时候，我要再去遍历所有的数据，依次叠加对吧，挨个再加一遍，最后算完输出结果，所以这两个的效率大家一看就能就能看出这个差别，对吧？啊所以呃，如果要是实际使用的过程当中呢，当然推荐大家是能用增量聚合就用增量聚合啊，那有同学可能想，那你这里这个全窗口函数还有什么用呢啊，有一些场景下，可能我们对于这个数据就没有办法去做增量聚合啊，就或者说。

你增量聚合得到那个那个数据，呃，没有太大的用，对吧？我最后还是得把所有的这个这个数收集齐了，然后去做一个处理操作，比方说什么呢？比方说我要排序对吧，或者说比方说我要输出的结果是当前所有数据里边百分之中位数对吧？25%分位，75%分位的那个数到底是什么？你说你这些数，其实你来一个数据就重新整合一遍，来一个数据重新整合一遍，呃，其实其实没用，对吧？你不能只保持一个结果，你必须把所有的数都保存下来，最后这个结果才是准确的。所以对于这种场景，你最后还是得用一个全窗口函数。另外呢，全窗口函数还有一个特点，就是它比这个增量聚合函数能够获取到的信息要多，它能够在上下文里边拿到当前窗口的信息，拿到当前啊，运行上下文里边状态的信息啊，所以他能做的操作会更多一些，它比较更底层，更灵活啊，所以就是如果有需求的话，我们也可以用它来做，那具体的这个举例，这个增量聚合函数的代表就是所谓的reduce function式和aggregate function，好啊，那包括那些简单的那些操作，大家看到那个sum max m对吧，它的底层其实也都是增量聚合的啊，那个我们就不说了，这里边主要说的就是这是一个函数类对吧？大家看到这是一个函数类，就是你如果什么时候调这个reduce function呢？这就是我们前面在。

API里边大家看到啊，在这个得到的这个window stream里面。这里边你假如说调一个reduce方法的话，你看这里边传的就是一个reduce方式，对吧？啊，那如果说这里边你去调一个aggregate agg啊aggregate方法的话，这里边诶常规的情况下，你就可以传的是一个aggregate方式啊，所以这里边它其实说的就是我们传就是给这个聚合操作啊，后面我们做这个窗口计算的时候，这个聚合操作里边传的这个。

函数类啊，这个东西叫做窗口函数，我们增量这算函数的话就是reduce function aggregate function，它是增量的啊，那同样还有就是还有一些窗口操作呢，是全窗口计算的，就是所有数据都收集齐了再算的，比方说什么呢？比方说大家看到有一个process process，对吧？Process我们知道它本身是一个process方式啊，之前给大家提到过，这是一个最底层的那个呃，接口process方式API，那它里边的这个调用要传的是一个process window function，呃，这就是收集齐了，所有数据都收集齐了，然后什么信息都能获取到对吧？就是什么？呃，想拿的这个上下文信息都能拿到，在这里边你可以做任何想想做的操作，它的灵活性更高一点啊。另外还有一个啊，大家看到这里边还有一个fold，对吧？啊，这个fold其实也是类似于我们前面那个，呃，像在那个key by之后，大家记得也有一个fold对吧？啊，这里边也是它被depated了，就是让你用。

Aggregate来来代替这个fold对吧，跟那边KY的定义是差不多的，所以我们不专门说这个，呃，Fold的这个你看fold这里边它可以传一个fold方式对吧？啊，这个我们就不专门去说了，那另外除了这个process function process window function之外呢，还有一个大家看到啊，有一个方法叫apply里边它可以直接传一个window function，这个就变成一个狭义的window function啊，这是一个具体的函数类对吧？这个window function也是一个全窗和函数啊，就是它也是把所有数据都收集齐了才做的操作啊，所以这里边大家可以就是记住这么几种，呃，不同的这个窗口函数的类型啊，全窗口函数里面代表就是这个window function和process window function，那增量聚合函数呢，就是reduce function和aggregate function，这样一说的话，大家可能觉得呃，还是稍微有点绕是吧，后面我们还是在代码里边具体给大家做操作，可能理解就会更加透彻一些，这里我们先把后面的这一部分。

部分都给大家讲完啊，然后下面还有一些可选的API，那这些可选API是什么意思呢？其实大家就看到了，我们在代码里边啊，如果说你做了这样的一个，不管是做什么啊，你看你做这个reduce这样的一个聚合也好，增量聚合也好，或者说你直接apply apply大家知道是应用嘛，对吧，这就不说是到底是增量聚合还是干什么，我就是想应用一个窗口函数而已啊，全窗口函数不管是什么，大家看最后得到的都是一个data stream，对吧，Apply process，你看得到的都是data stream。包括max m sun，对吧，都是绕一圈之后，由这个window stream又得到了一个data stream啊，所以整个的这个计算过程大家就发现了啊，所谓的data stream API，我们就说它的核心是data stream。

然后呢啊，前面我们讲过，可以去做各种各样的转换，然后这种转换基本上都是成对出现的啊，为什么成对出现呢？就是你先出去，然后最后还要绕回来，对不对啊，之前大家比较熟悉的是我可以先KK之后得到的是一个K的我这个简写了啊。然后之后我再做一些聚合操作，对吧，比方说some m max，或者说更加一般化的reduce，最后就又得到了data，啊，这个聚合，这是连续不断的，大家发现这是基于这个无介流的这样的分组之后的聚合，对吧？那现在呢，如果常用的我们是要做有界流，那怎么办呢？哎，那在这个K的stream基础上再去开窗window，这样得到的就是一个window stream，然后这个window stream呢，又可以再去做转换，哎，这里边又是各种各样的聚合，对吧？Some reduce aggregate。

最后还是得到一个data stream，所以就是转一圈，最后还是要回到它来，对吧？还是基于data stream去做各种各样不同的转换，所以这里大家就看到了，在这个window操作里边呢，这个window的定义啊，就是这个window a sign，以及后面我们说的，你随便做一个reduce也好，或者做一个aggregate也好，这两步操作是必不可少的，就是你定义window长什么样，这个数据来了之后分配到哪个window去，然后呢，呃，再定义一个到时候的，呃，就是window关闭的时候到底该怎么做计算，这两步必不可少。那另外还有一些呢可选，这个可选的就是这里边列出来的API，比方说trigger触发器，前面我们看到了，对吧？Global window里边count window啊，你就是基于global window去自定义一个触发器，它的含义就是定义这个window到底什么时候关闭，然后什么时候触发计算输出结果，这就是trigger的用途，然后另外呢，还可以定义对吧。

啊X，我们叫移除器，它主要就是定义啊，我哪些数据在做计算的时候从这个窗口里边移除出去啊，就是做计算的时候不要考虑这些，有点像一个filter一样，对吧，类似于这样的一个功能，另外还有几个比较有意思的功能啊，这里边大家看到还有一个叫什么呢？Alo lateness，就是当前在flink里边，我可以允许开启这个之后啊，就可以允许处理迟到数据了，之前我们不是说当前这个分布式架构里边有可能会出现这个数据的乱序嘛，有可能会出现本来先发生的这个事件，对吧，先产生的数据到后面我们处理的时候，它它到后面才来吗？那如果说我们这个窗口要关闭的时候，他那数据还没到姗姗来迟怎么办呢？诶，我可以等他一会儿对吧，这里边lo listens里边可以传一个时间，一个time，就像我们前面定义的那个time一样，你比方说一个time.MINUTES1，那就是允许处理一分钟之内的迟到数据，对吧，多等一。

分钟啊，然后另外呢，还有一个叫做side output late data，这又是什么含义呢？啊，这个说的是把迟到的数据放到测输出流啊，这个一般情况这是配合起来用的啊，这就是什么？大家想你如果允许处理迟到数据的话，那你说我这个迟到数据我到底该等多久呢？你等一分钟也是等，等十分钟也是等，对吧，到底等多久才能保证我最后的结果一定正确呢？这就没准了啊，所以弗link的处理是因为大家想到你如果说我这里边的这个数据一直要让他等待的话，是不是当前我窗口里边的那个，呃，数据也好，或者说聚合出来的状态也好，窗窗口的各种这个上下文的一些内容啊，一些状态是不是就都得保持在这儿不能释放啊，这都会占据内存资源对吧？哎，所以说你如果要是这个给的时间太长的话，可能对我们那个内存的压力会比较大，你不能无限等下去，所以一般情况我这里边就是大概给一个差不多的时间，对吧，我觉得等一分。

分钟大部分应该来了，那假如说还有少量漏网之鱼怎么办呢？哎，没关系，如果还有漏网之鱼的话，我把它扔到测输出流里边去哈，啊，就是这一般是配合使用的啊，至于测输出流到底是什么，后边我们给大家讲那个process function的时候啊，再给大家详细讲这一部分啊，那这个你如果要已经把它放到测输测输出流之后，那之后怎么把它再拿出来呢？哎，那就是最后我们已经完成这个window计算操作之后，得到的是一个data stream对吧？哎，那你基于这个结果的这个data stream，然后再去get side output就可以获取测试出流了，这是其他一些可选的API啊，所以最后我们可以给大家再来做一个总结，就是所有window API有哪些呢？能调的就这些啊，就是一般情况我们调用的时候是先做KBY分组之后调window去去得到一个key window的stream，然后去调这个window相关的操作，对吧？啊，那还有另外一种方式，就是不KBY。

不分组直接WINDOW2，基于一个data stream直接WINDOW2，然后后边操作几乎完全一样啊，只是这里边有一个window和WINDOW2的差别而已啊，这里边官网其实是不推荐大家直接用window all啊，为什么呢？因为如果要是直接用window or的话，Flink底层会把我们所有的数据都直接发送到同一个分区里边去去做，然后再做这个窗口的这个分筒，再去做计算，所以大家想这就相当于什么，相当于并行度就直接变一了，对不对啊，你就相当于整个任务就没有并行了，所以说这个对性能是有有有损失的啊，就如果说呃不是必要就非得这么干的话，那我们尽量还是不要这样WINDOW2对吧，我们还是直接K分组之后再去开窗。然后呢，大家看到必须的两个操作，一个是window定义a sign对吧，然后另外还有一个就是定义window function啊，就是这里边就可以reduce aggregate这两个里边传的是增量聚合函数，或者这个呃，A food被弃用了对吧，或者是apply process，这里边传的是全冲构函数啊，大家大概知道怎么用就可以了，然后再注意啊，大家看啊，在它俩之间我可以再去定义各种各样的这个可选的这个这些东西啊，义这个trigger触发器移除器对吧，允许这个处理迟到数据啊，另外呢，还可以把它扔到这个测输出流里边去啊，那最后大家注意，你如果扔到测殊出流，最后怎么拿出来呢？大家看是reduce，就是做完这个窗口操作之后得到了data stream，然后基于这个data stream，再去调get set output，把它拿出来。

是这样的一种操作，好啊，那大家就可以把这个window的操作好好再总结一下了啊。