只有单页面的网站怎么做seo,网站被很多公司抄袭,ysl 官网中国,供求信息网站建设报价窗口#xff08;Window#xff09;
窗口是处理无限流的核心。 窗口将流分割成有限大小的“桶”#xff0c;我们可以计算窗口中的数据。
窗口程序一般有键控流#xff08;keyed streams#xff09;的窗口程序 和 非键控流#xff08;non-keyed streams#xff09;的窗口… 窗口Window
窗口是处理无限流的核心。 窗口将流分割成有限大小的“桶”我们可以计算窗口中的数据。
窗口程序一般有键控流keyed streams的窗口程序 和 非键控流non-keyed streams的窗口程序。
监控流keyed streams的窗口程序
stream.keyBy(...) - keyed versus non-keyed windows.window(...) - required: assigner[.trigger(...)] - optional: trigger (else default trigger)[.evictor(...)] - optional: evictor (else no evictor)[.allowedLateness(...)] - optional: lateness (else zero)[.sideOutputLateData(...)] - optional: output tag (else no side output for late data).reduce/aggregate/apply() - required: function[.getSideOutput(...)] - optional: output tag
非监控流non-keyed streams的窗口程序
stream.windowAll(...) - required: assigner[.trigger(...)] - optional: trigger (else default trigger)[.evictor(...)] - optional: evictor (else no evictor)[.allowedLateness(...)] - optional: lateness (else zero)[.sideOutputLateData(...)] - optional: output tag (else no side output for late data).reduce/aggregate/apply() - required: function[.getSideOutput(...)] - optional: output tag 窗口生命周期
窗口的创建一旦应该属于该窗口的第一个元素到达就会创建一个窗口。
窗口的删除当时间事件或处理时间超过其结束时间戳加上用户指定的允许延迟时该窗口将被完全删除。 Flink 保证仅删除基于时间的窗口而不保证删除其他类型例如 全局窗口。
例如采用基于事件时间的窗口策略每 5 分钟创建一次非重叠或翻滚窗口并且允许延迟 1 分钟如12:00 到 12:05 是一个窗口周期当第一个时间戳落入该区间的元素到达时Flink创建该窗口当水印经过 12:06 时间戳时Flink将删除这个窗口。
此外每个窗口都会附加一个触发器和一个函数ProcessWindowFunction、ReduceFunction 或 AggregateFunction。 该函数将包含要应用于窗口内容的计算而触发器指定窗口被视为准备好应用函数的条件。 触发策略可能类似于“当窗口中的元素数量超过 4 时”或“当水印经过窗口末尾时”。 触发器还可以决定在创建和删除窗口之间的任何时间清除窗口的内容。 在这种情况下清除仅指窗口中的元素而不是窗口元数据。 这意味着新数据仍然可以添加到该窗口中。
除此之外还可以指定一个 Evictor它将能够在触发器触发后以及应用函数之前和/或之后从窗口中删除元素。 键控窗口与非键控窗口
流数据 使用 keyBy(...) 会将无限流拆分为逻辑键控流。 如果未调用 keyBy(...)则是非键控流。
对于键控流传入事件的任何属性都可以用作键。 拥有键控流将允许多个任务并行执行窗口计算因为每个逻辑键控流都可以独立于其余流进行处理。 引用相同键的所有元素将被发送到相同的并行任务。
对于非键控流原始流不会被分割成多个逻辑流并且所有窗口逻辑将由单个任务执行即并行度为 1。 窗口分配器Window Assigners
在指定了流是键控的还是非键控的后下一步是定义窗口分配器。
窗口分配器决定了如何将元素分配到各个窗口中。
通常通过在键控流中使用window()方法或在非键控流中使用windowAll()方法并指定所需的WindowAssigner来实现。
窗口分配器负责将每个传入的元素分配到一个或多个窗口中。Flink为最常见的用例提供了预定义的窗口分配器即滚动窗口、滑动窗口、会话窗口和全局窗口。
此外用户还可以通过扩展WindowAssigner类来实现自定义窗口分配器。
除了全局窗口外所有内置窗口分配器都是基于时间可以是处理时间或事件时间来分配元素的。
基于时间的窗口有一个开始时间戳包含和一个结束时间戳不包含这两个时间戳共同描述了窗口的大小。在代码中Flink使用TimeWindow类来处理基于时间的窗口该类提供了查询开始和结束时间戳的方法以及一个返回给定窗口允许的最大时间戳的额外方法。 滚动窗口
滚动窗口分配器将每个元素分配到一个具有指定窗口大小的窗口中。滚动窗口具有固定的大小并且不会重叠。
每个窗口的有开始和结束时间戳随着时间的推移元素会根据它们到达的时间被分配到相应的窗口中。一旦一个窗口的时间范围结束该窗口就会被关闭并且其中的元素会被处理例如进行聚合操作。然后一个新的窗口会立即开始并等待新的元素被分配进来。
这种滚动窗口的分配方式适用于那些需要定期分析固定时间段内数据的场景比如每5分钟统计一次网站访问量。由于窗口之间不重叠每个元素只会被分配到一个窗口中这有助于简化数据处理逻辑。
请注意使用滚动窗口时需要确保数据流中的时间戳是准确且有序的如果数据流的时间戳不准确或乱序可能会导致窗口分配错误从而影响分析结果的准确性。 Java代码示例
DataStreamT input ...;// tumbling event-time windows
input.keyBy(key selector).window(TumblingEventTimeWindows.of(Time.seconds(5))).windowed transformation(window function);// tumbling processing-time windows
input.keyBy(key selector).window(TumblingProcessingTimeWindows.of(Time.seconds(5))).windowed transformation(window function);// daily tumbling event-time windows offset by -8 hours.
input.keyBy(key selector).window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8))).windowed transformation(window function); 时间间隔可以使用Time.milliseconds(x)、Time.seconds(x)、Time.minutes(x)等函数来指定。
正如上一个例子所示滚动窗口分配器还接受一个可选的偏移参数该参数可用于改变窗口的对齐方式。例如如果不设置偏移量则每小时的滚动窗口会与纪元时间对齐即会得到如1:00:00.000 - 1:59:59.999、2:00:00.000 - 2:59:59.999等这样的窗口。如果想要改变对齐方式你可以设置一个偏移量。例如如果设置了一个15分钟的偏移量那么将得到1:15:00.000 - 2:14:59.999、2:15:00.000 - 3:14:59.999等这样的窗口。偏移量的一个重要用例是调整窗口以适应除UTC-0时区之外的其他时区。例如在中国需要指定一个偏移量Time.hours(-8)。
通过使用偏移量可以确保窗口的边界与特定时区的时间边界对齐这在处理具有时区特定需求的数据时非常有用。这样无论Flink作业在哪个时区运行都可以确保数据按照期望的时间范围进行窗口化从而得到准确的分析结果。
请注意偏移量仅影响窗口的起始时间窗口的大小即时间间隔仍然保持不变。 滑动窗口
滑动窗口分配器将元素分配到固定长度的窗口中。与滚动窗口分配器类似窗口的大小由窗口大小参数配置。而一个额外的窗口滑动参数控制着滑动窗口启动的频率。因此如果滑动间隔小于窗口大小滑动窗口可以重叠。在这种情况下元素会被分配到多个窗口中。
例如你可以设置大小为10分钟的窗口每5分钟滑动一次。这样你每5分钟就会得到一个包含过去10分钟内到达的事件的窗口如以下图表所示。
在这个图表中可以看到窗口如何随着时间的推移而滑动。每个窗口包含了一定时间段内的所有事件这些事件根据它们到达的时间被分配到相应的窗口中。由于滑动间隔小于窗口大小所以窗口之间存在重叠部分。这种重叠使得滑动窗口能够在时间上提供更加平滑的覆盖适用于需要持续追踪一段时间内事件变化的场景如流量分析、温度监测等。
使用滑动窗口时需要仔细考虑窗口大小和滑动间隔的设置以确保能够满足分析需求。如果窗口太大可能会导致分析结果过于粗糙如果窗口太小或滑动间隔太大可能会丢失一些重要的细节信息。因此在实际应用中可能需要根据数据的特性和分析目标来调整这些参数。 Java代码示例
DataStreamT input ...;// sliding event-time windows
input.keyBy(key selector).window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5))).windowed transformation(window function);// sliding processing-time windows
input.keyBy(key selector).window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5))).windowed transformation(window function);// sliding processing-time windows offset by -8 hours
input.keyBy(key selector).window(SlidingProcessingTimeWindows.of(Time.hours(12), Time.hours(1), Time.hours(-8))).windowed transformation(window function); 会话窗口
会话窗口分配器根据活动会话将元素分组。与滚动窗口和滑动窗口不同会话窗口不会重叠也没有固定的开始和结束时间。相反会话窗口在没有接收到元素一段时间后关闭即当发生一段时间的不活跃期时。会话窗口分配器可以配置为使用静态会话间隔或者使用会话间隔提取函数来定义不活跃期的时长。当这个时段结束时当前会话关闭后续的元素将被分配到新的会话窗口中。
使用会话窗口在处理具有不同活跃期的数据流时非常有用比如用户会话、网页浏览事件等。由于会话窗口是根据实际活动来定义的因此它们能够更自然地反映数据的实际模式。
配置会话窗口时需要确定一个合适的会话间隔。这个间隔应该足够长以便能够捕捉到会话中的活动但又不能太长以免将不同的会话错误地合并在一起。如果会话模式变化很大可以考虑使用会话间隔提取函数以便根据每个会话的具体特性动态地调整间隔时长。
通过使用会话窗口可以更好地理解和分析数据中的活动模式从而得到更准确的业务洞察。 Java代码示例
DataStreamT input ...;// event-time session windows with static gap
input.keyBy(key selector).window(EventTimeSessionWindows.withGap(Time.minutes(10))).windowed transformation(window function);// event-time session windows with dynamic gap
input.keyBy(key selector).window(EventTimeSessionWindows.withDynamicGap((element) - {// determine and return session gap})).windowed transformation(window function);// processing-time session windows with static gap
input.keyBy(key selector).window(ProcessingTimeSessionWindows.withGap(Time.minutes(10))).windowed transformation(window function);// processing-time session windows with dynamic gap
input.keyBy(key selector).window(ProcessingTimeSessionWindows.withDynamicGap((element) - {// determine and return session gap})).windowed transformation(window function); 全局窗口
全局窗口分配器将所有具有相同键的元素分配到同一个全局窗口中。这种窗口方案仅在指定了自定义触发器时才有用。否则将不会执行任何计算因为全局窗口没有自然的结束点来处理聚合的元素。
全局窗口提供了一个灵活的框架允许你根据自定义逻辑来定义窗口的结束。由于全局窗口不会根据时间或其他标准自动关闭因此需要提供一个触发器来确定何时应该关闭窗口并触发计算。
在使用全局窗口时你需要仔细考虑如何定义触发器以确保窗口在适当的时候关闭。触发器可以根据元素的数量、特定的时间条件、外部事件等因素来触发窗口的关闭。通过自定义触发器实现复杂的窗口逻辑以满足特定的业务需求。
需要注意的是全局窗口可能会导致大量的数据在内存中积累直到触发器被触发为止。因此在使用全局窗口时需要确保系统具有足够的内存和处理能力来处理可能的大数据量。此外需要仔细设计触发器的逻辑以避免出现窗口长时间不关闭或频繁关闭的情况这可能会影响系统的性能和稳定性。 Java代码示例
DataStreamT input ...;input.keyBy(key selector).window(GlobalWindows.create()).windowed transformation(window function); 窗口函数Window Functions
ReduceFunction
ReduceFunction 指定如何组合输入中的两个元素以生成相同类型的输出元素。 Flink 使用ReduceFunction 增量聚合窗口的元素。
Java示例代码
DataStreamTuple2String, Long input ...;input.keyBy(key selector).window(window assigner).reduce(new ReduceFunctionTuple2String, Long() {public Tuple2String, Long reduce(Tuple2String, Long v1, Tuple2String, Long v2) {return new Tuple2(v1.f0, v1.f1 v2.f1);}}); AggregateFunction
AggregateFunction 是ReduceFunction 的通用版本具有三种类型输入类型(IN)、累加器类型(ACC) 和输出类型(OUT)。 输入类型是输入流中元素的类型AggregateFunction 具有一种将一个输入元素添加到累加器的方法。 该接口还具有用于创建初始累加器、将两个累加器合并为一个累加器以及从累加器提取输出OUT 类型的方法。 我们将在下面的示例中看到它是如何工作的。 与ReduceFunction 相同Flink 将在窗口的输入元素到达时增量聚合它们。
Java代码示例 /*** The accumulator is used to keep a running sum and a count. The {code getResult} method* computes the average.*/
private static class AverageAggregateimplements AggregateFunctionTuple2String, Long, Tuple2Long, Long, Double {Overridepublic Tuple2Long, Long createAccumulator() {return new Tuple2(0L, 0L);}Overridepublic Tuple2Long, Long add(Tuple2String, Long value, Tuple2Long, Long accumulator) {return new Tuple2(accumulator.f0 value.f1, accumulator.f1 1L);}Overridepublic Double getResult(Tuple2Long, Long accumulator) {return ((double) accumulator.f0) / accumulator.f1;}Overridepublic Tuple2Long, Long merge(Tuple2Long, Long a, Tuple2Long, Long b) {return new Tuple2(a.f0 b.f0, a.f1 b.f1);}
}DataStreamTuple2String, Long input ...;input.keyBy(key selector).window(window assigner).aggregate(new AverageAggregate()); ProcessWindowFunction
ProcessWindowFunction 获取一个包含窗口所有元素的 Iterable以及一个可以访问时间和状态信息的 Context 对象这使其能够提供比其他窗口函数更大的灵活性。 这是以性能和资源消耗为代价的因为元素不能增量聚合而是需要在内部缓冲直到窗口被认为准备好进行处理。
源码如下
public abstract class ProcessWindowFunctionIN, OUT, KEY, W extends Window implements Function {/*** Evaluates the window and outputs none or several elements.** param key The key for which this window is evaluated.* param context The context in which the window is being evaluated.* param elements The elements in the window being evaluated.* param out A collector for emitting elements.** throws Exception The function may throw exceptions to fail the program and trigger recovery.*/public abstract void process(KEY key,Context context,IterableIN elements,CollectorOUT out) throws Exception;/*** Deletes any state in the {code Context} when the Window expires (the watermark passes its* {code maxTimestamp} {code allowedLateness}).** param context The context to which the window is being evaluated* throws Exception The function may throw exceptions to fail the program and trigger recovery.*/public void clear(Context context) throws Exception {}/*** The context holding window metadata.*/public abstract class Context implements java.io.Serializable {/*** Returns the window that is being evaluated.*/public abstract W window();/** Returns the current processing time. */public abstract long currentProcessingTime();/** Returns the current event-time watermark. */public abstract long currentWatermark();/*** State accessor for per-key and per-window state.** pbNOTE:/bIf you use per-window state you have to ensure that you clean it up* by implementing {link ProcessWindowFunction#clear(Context)}.*/public abstract KeyedStateStore windowState();/*** State accessor for per-key global state.*/public abstract KeyedStateStore globalState();}}
key 参数是通过为 keyBy() 调用指定的 KeySelector 提取的密钥。 如果是元组索引键或字符串字段引用则此键类型始终为 Tuple必须手动将其转换为正确大小的元组才能提取键字段。 Java代码示例
DataStreamTuple2String, Long input ...;input.keyBy(t - t.f0).window(TumblingEventTimeWindows.of(Time.minutes(5))).process(new MyProcessWindowFunction());/* ... */public class MyProcessWindowFunction extends ProcessWindowFunctionTuple2String, Long, String, String, TimeWindow {Overridepublic void process(String key, Context context, IterableTuple2String, Long input, CollectorString out) {long count 0;for (Tuple2String, Long in: input) {count;}out.collect(Window: context.window() count: count);}
}
该示例显示了一个对窗口中的元素进行计数的 ProcessWindowFunction。此外窗口函数还将有关窗口的信息添加到输出中。
请注意使用 ProcessWindowFunction 进行简单聚合例如计数效率相当低。 带有增量聚合的ProcessWindowFunction
ProcessWindowFunction 可以与ReduceFunction 或AggregateFunction 结合使用以在元素到达窗口时增量聚合元素。 当窗口关闭时将向 ProcessWindowFunction 提供聚合结果。 这允许它增量计算窗口同时可以访问 ProcessWindowFunction 的附加窗口元信息。 还可以使用旧版 WindowFunction 而不是 ProcessWindowFunction 进行增量窗口聚合。 ProcessWindowFunction 结合 ReduceFunction
以下示例展示了如何将增量式ReduceFunction 与ProcessWindowFunction 结合起来以返回窗口中的最小事件以及窗口的开始时间。
Java代码示例
DataStreamSensorReading input ...;input.keyBy(key selector).window(window assigner).reduce(new MyReduceFunction(), new MyProcessWindowFunction());// Function definitionsprivate static class MyReduceFunction implements ReduceFunctionSensorReading {public SensorReading reduce(SensorReading r1, SensorReading r2) {return r1.value() r2.value() ? r2 : r1;}
}private static class MyProcessWindowFunctionextends ProcessWindowFunctionSensorReading, Tuple2Long, SensorReading, String, TimeWindow {public void process(String key,Context context,IterableSensorReading minReadings,CollectorTuple2Long, SensorReading out) {SensorReading min minReadings.iterator().next();out.collect(new Tuple2Long, SensorReading(context.window().getStart(), min));}
}
ProcessWindowFunction 结合 AggregateFunction
以下示例展示了如何将增量 AggregateFunction 与 ProcessWindowFunction 结合起来计算平均值并发出键和窗口的平均值。 Java代码示例
DataStreamTuple2String, Long input ...;input.keyBy(key selector).window(window assigner).aggregate(new AverageAggregate(), new MyProcessWindowFunction());// Function definitions/*** The accumulator is used to keep a running sum and a count. The {code getResult} method* computes the average.*/
private static class AverageAggregateimplements AggregateFunctionTuple2String, Long, Tuple2Long, Long, Double {Overridepublic Tuple2Long, Long createAccumulator() {return new Tuple2(0L, 0L);}Overridepublic Tuple2Long, Long add(Tuple2String, Long value, Tuple2Long, Long accumulator) {return new Tuple2(accumulator.f0 value.f1, accumulator.f1 1L);}Overridepublic Double getResult(Tuple2Long, Long accumulator) {return ((double) accumulator.f0) / accumulator.f1;}Overridepublic Tuple2Long, Long merge(Tuple2Long, Long a, Tuple2Long, Long b) {return new Tuple2(a.f0 b.f0, a.f1 b.f1);}
}private static class MyProcessWindowFunctionextends ProcessWindowFunctionDouble, Tuple2String, Double, String, TimeWindow {public void process(String key,Context context,IterableDouble averages,CollectorTuple2String, Double out) {Double average averages.iterator().next();out.collect(new Tuple2(key, average));}
}在ProcessWindowFunction中使用每个窗口的状态
除了访问键控状态任何丰富的函数都可以之外ProcessWindowFunction 还可以使用作用域为函数当前正在处理的窗口的键控状态。 在这种情况下了解每个窗口状态所指的窗口是什么非常重要。 涉及不同的“窗口”
指定窗口操作时定义的窗口这可能是 1 小时的滚动窗口或滑动 1 小时的 2 小时滑动窗口。 给定键的已定义窗口的实际实例对于用户 ID xyz这可能是从 12:00 到 13:00 的时间窗口。 这是基于窗口定义的并且根据作业当前正在处理的键的数量以及事件所属的时隙将会有许多窗口。 每个窗口的状态与这两者中的后者相关。 这意味着如果我们处理 1000 个不同键的事件并且所有这些键的事件当前都落入 [12:00, 13:00) 时间窗口那么将有 1000 个窗口实例每个实例都有自己的键控每个窗口状态。
process() 调用接收的 Context 对象上有两个方法允许访问两种类型的状态
globalState()它允许访问不在窗口范围内的键控状态 windowState()它允许访问范围也在窗口内的键控状态 如果预计同一窗口会多次触发则此功能非常有用当对迟到的数据进行延迟触发或当有一个进行推测性早期触发的自定义触发器时可能会发生这种情况。 在这种情况下将存储有关先前触发或每个窗口状态的触发次数的信息。
使用窗口状态时清除窗口时清除该状态也很重要。 这应该发生在clear()方法中。 WindowFunction旧版
在某些可以使用 ProcessWindowFunction 的地方也可以使用 WindowFunction。 这是 ProcessWindowFunction 的旧版本提供的上下文信息较少并且没有一些高级功能例如每个窗口的键控状态。 该接口将在某个时候被弃用。
源码如下
public interface WindowFunctionIN, OUT, KEY, W extends Window extends Function, Serializable {/*** Evaluates the window and outputs none or several elements.** param key The key for which this window is evaluated.* param window The window that is being evaluated.* param input The elements in the window being evaluated.* param out A collector for emitting elements.** throws Exception The function may throw exceptions to fail the program and trigger recovery.*/void apply(KEY key, W window, IterableIN input, CollectorOUT out) throws Exception;
}
Java代码示例
DataStreamTuple2String, Long input ...;input.keyBy(key selector).window(window assigner).apply(new MyWindowFunction()); 触发器Triggers
触发器确定窗口由窗口分配器形成何时准备好由窗口函数处理。 每个 WindowAssigner 都带有一个默认的触发器。 如果默认触发器不能满足业务需求可以使用trigger(...)指定自定义触发器。
触发器接口有五种方法允许触发器对不同的事件做出反应
每个添加到窗口的元素都会调用 onElement() 方法。 当注册的事件时间计时器触发时将调用 onEventTime() 方法。 当注册的处理时间计时器触发时将调用 onProcessingTime() 方法。 onMerge() 方法与有状态触发器相关并在两个触发器相应的窗口合并时合并两个触发器的状态例如 使用会话窗口时。 最后clear() 方法删除相应窗口时执行所需的任何操作。 上述方法需要注意的两点是
1.前三个决定如何通过返回 TriggerResult 来处理其调用事件。 该操作可以是以下操作之一
CONTINUE什么也不做FIRE触发计算PURGE清除窗口中的元素FIRE_AND_PURGE触发计算并随后清除窗口中的元素 2.这些方法中的任何一个都可以用于注册处理或事件时间计时器以用于将来的操作。 触发与清除
一旦触发器确定窗口已准备好进行处理它就会触发即返回 FIRE 或 FIRE_AND_PURGE。 这是窗口操作符发出当前窗口结果的信号。 给定一个带有 ProcessWindowFunction 的窗口所有元素都会传递给 ProcessWindowFunction可能在将它们传递给逐出器之后。 带有ReduceFunction 或AggregateFunction 的Windows 只是简单地发出它们急切的聚合结果。
当触发器触发时它可以是 FIRE 或 FIRE_AND_PURGE。 FIRE 保留窗口的内容而 FIRE_AND_PURGE 则删除其内容。 默认情况下预实现的触发器只是简单地触发而不清除窗口状态。
清除只会删除窗口的内容并完整保留有关窗口的任何潜在元信息和任何触发器状态。 窗口分配器的默认触发器
WindowAssigner 的默认触发器适用于许多用例。 例如所有事件时间窗口分配器都有一个 EventTimeTrigger 作为默认触发器。 一旦水印经过窗口末尾就会触发此触发器。
GlobalWindow 的默认触发器是 NeverTrigger它从不触发。 因此在使用 GlobalWindow 时始终必须定义自定义触发器。
通过使用trigger()指定触发器将覆盖WindowAssigner的默认触发器。 例如如果为 TumblingEventTimeWindows 指定 CountTrigger将不再根据时间进度而仅根据计数来触发窗口。 如果想根据时间和计数做出反应则必须编写自己的自定义触发器。 内置触发器和自定义触发器
Flink 带有一些内置触发器。
已经提到的EventTimeTrigger 根据水印测量的事件时间进度触发。ProcessingTimeTrigger 根据处理时间触发。一旦窗口中的元素数量超过给定限制CountTrigger 就会触发。PurgingTrigger 将另一个触发器作为参数并将其转换为清除触发器。
如果需要实现自定义触发器应该查看Trigger抽象类。 请注意该 API 仍在不断发展并且可能会在 Flink 的未来版本中发生变化。 清除器Evictors
除了 WindowAssigner 和 Trigger 之外Flink 的窗口模型还允许指定可选的 Evictor。 这可以使用 evictor(...) 方法来完成如本文开头所示。 清除器能够在触发器触发之后以及应用窗口函数之前和/或之后从窗口中删除元素。 为此Evictor 接口有两种方法
/*** Optionally evicts elements. Called before windowing function.** param elements The elements currently in the pane.* param size The current number of elements in the pane.* param window The {link Window}* param evictorContext The context for the Evictor*/
void evictBefore(IterableTimestampedValueT elements, int size, W window, EvictorContext evictorContext);/*** Optionally evicts elements. Called after windowing function.** param elements The elements currently in the pane.* param size The current number of elements in the pane.* param window The {link Window}* param evictorContext The context for the Evictor*/
void evictAfter(IterableTimestampedValueT elements, int size, W window, EvictorContext evictorContext);
evictBefore() 包含在窗口函数之前应用的清除逻辑而 evictAfter() 包含在窗口函数之后应用的清除逻辑。 在应用窗口函数之前被清除的元素将不会被窗口函数处理。
Flink 附带了三个预先实现的清除器。 这些都是
CountEvictor保留窗口中用户指定数量的元素并丢弃窗口缓冲区开头的剩余元素。DeltaEvictor采用 DeltaFunction 和阈值计算窗口缓冲区中最后一个元素与其余每个元素之间的增量并删除增量大于或等于阈值的元素。TimeEvictor以毫秒为单位的间隔作为参数对于给定的窗口它在其元素中查找最大时间戳 max_ts 并删除时间戳小于 max_ts - interval的所有元素。
默认情况下所有预先实现的清除器都在窗口函数之前应用其逻辑。
指定清除器可以防止任何预聚合因为在应用计算之前必须将窗口的所有元素传递给清除器。这意味着带有清除器的窗口将创建更多的状态。
Flink 不保证窗口内元素的顺序。这意味着虽然清除器可以从窗口的开头删除元素但这些元素不一定是最先或最后到达的元素。 允许延迟
当使用事件时间窗口时元素可能会延迟到达即 Flink 用于跟踪事件时间进度的水印已经超过了元素所属窗口的结束时间戳。
默认情况下当水印超过窗口末尾时迟到的元素将被删除。 然而Flink 允许为窗口操作符指定允许的最大延迟时间。 允许的延迟指定元素在被删除之前可以延迟的时间其默认值为 0。在水印经过窗口末尾但在经过窗口末尾之前到达的元素加上允许的延迟 仍然添加到窗口中。 根据所使用的触发器迟到但未删除的元素可能会导致窗口再次触发。 EventTimeTrigger 就是这种情况。
为了实现这一点Flink 会保留窗口的状态直到其允许的延迟到期。 一旦发生这种情况Flink 将删除窗口并删除其状态。
默认情况下允许的延迟设置为 0。也就是说在水印之后到达的元素将被丢弃。
Java代码示例
DataStreamT input ...;input.keyBy(key selector).window(window assigner).allowedLateness(time).windowed transformation(window function);
使用 GlobalWindows 窗口分配器时不会将任何数据视为迟到因为全局窗口的结束时间戳为 Long.MAX_VALUE。 将延迟数据作为侧输出
使用 Flink 的侧输出功能可以获得最新丢弃的数据流。
首先需要指定要在窗口流上使用 sideOutputLateData(OutputTag) 获取延迟数据。 然后可以获得窗口操作结果的侧输出流
Java代码示例
final OutputTagT lateOutputTag new OutputTagT(late-data){};DataStreamT input ...;SingleOutputStreamOperatorT result input.keyBy(key selector).window(window assigner).allowedLateness(time).sideOutputLateData(lateOutputTag).windowed transformation(window function);DataStreamT lateStream result.getSideOutput(lateOutputTag); 关于延迟元素的考虑
当指定允许的延迟大于 0 时窗口及其内容将在水印经过窗口末尾后保留。 在这些情况下当一个迟到但未删除的元素到达时它可能会触发窗口的另一次触发。 这些触发称为延迟触发因为它们是由延迟事件触发的与窗口第一次触发的主触发相反。 在会话窗口的情况下延迟触发可能会进一步导致窗口合并因为它们可能会“弥合”两个预先存在的未合并窗口之间的间隙。
延迟触发发出的元素应被视为先前计算的更新结果即您的数据流将包含同一计算的多个结果。 根据应用程序需要考虑这些重复的结果或删除它们的重复项。 处理窗口结果
窗口操作的结果又是一个数据流结果元素中不会保留有关窗口操作的信息因此如果想保留有关窗口的元信息则必须在 ProcessWindowFunction 的结果元素中手动编码该信息。 在结果元素上设置的唯一相关信息是元素时间戳。 这被设置为已处理窗口的最大允许时间戳即结束时间戳 - 1因为窗口结束时间戳是独占的。 请注意这对于事件时间窗口和处理时间窗口都是如此。 即在窗口操作之后元素始终具有时间戳但这可以是事件时间时间戳或处理时间时间戳。 对于处理时间窗口这没有特殊含义但对于事件时间窗口这与水印与窗口的交互方式一起启用了具有相同窗口大小的连续窗口操作。 水印与窗口的交互
当水印到达窗口操作符时会触发两件事
水印触发最大时间戳即结束时间戳 - 1小于新水印的所有窗口的计算水印按原样转发到下游操作
直观地说一旦下游操作收到该水印水印就会“清除”所有在下游操作中被视为后期的窗口。 连续的窗口操作
如前所述计算窗口结果的时间戳的方式以及水印与窗口交互的方式允许将连续的窗口操作串在一起。 当想要执行两个连续的窗口操作其中想要使用不同的键但仍希望来自同一上游窗口的元素最终出现在同一下游窗口中时这可能很有用。
Java代码示例
DataStreamInteger input ...;DataStreamInteger resultsPerKey input.keyBy(key selector).window(TumblingEventTimeWindows.of(Time.seconds(5))).reduce(new Summer());DataStreamInteger globalResults resultsPerKey.windowAll(TumblingEventTimeWindows.of(Time.seconds(5))).process(new TopKWindowFunction());
在此示例中第一个操作的时间窗口 [0, 5) 的结果也将在后续窗口操作中的时间窗口 [0, 5) 中结束。 这允许计算每个键的总和然后在第二次操作中计算同一窗口内的前 k 个元素。 有用的状态大小考虑
窗口可以在很长一段时间内例如几天、几周或几个月进行定义因此会积累非常大的状态。 在估计窗口计算的存储需求时需要记住一些规则
Flink 为每个元素所属的窗口创建一个副本。 鉴于此翻滚窗口保留每个元素的一份副本一个元素恰好属于一个窗口除非它被延迟删除。 相反滑动窗口为每个元素创建多个元素如“窗口分配器”部分中所述。 因此大小为 1 天、滑动 1 秒的滑动窗口可能不是一个好主意。
ReduceFunction 和 AggregateFunction 可以显着降低存储需求因为它们急切地聚合元素并且每个窗口仅存储一个值。 相反仅使用 ProcessWindowFunction 就需要累积所有元素。
使用清除器可以防止任何预聚合因为在应用计算之前窗口的所有元素都必须通过清除器。 完成enjoy it! 文章转载自： http://www.morning.mbpfk.cn.gov.cn.mbpfk.cn http://www.morning.hqbk.cn.gov.cn.hqbk.cn http://www.morning.qbjgw.cn.gov.cn.qbjgw.cn http://www.morning.srxhd.cn.gov.cn.srxhd.cn http://www.morning.rjznm.cn.gov.cn.rjznm.cn http://www.morning.hlmkx.cn.gov.cn.hlmkx.cn http://www.morning.cxryx.cn.gov.cn.cxryx.cn http://www.morning.jfxth.cn.gov.cn.jfxth.cn http://www.morning.bhpsz.cn.gov.cn.bhpsz.cn http://www.morning.ygkk.cn.gov.cn.ygkk.cn http://www.morning.bchgl.cn.gov.cn.bchgl.cn http://www.morning.xppj.cn.gov.cn.xppj.cn http://www.morning.qhtlq.cn.gov.cn.qhtlq.cn http://www.morning.wflsk.cn.gov.cn.wflsk.cn http://www.morning.lhygbh.com.gov.cn.lhygbh.com http://www.morning.rycbz.cn.gov.cn.rycbz.cn http://www.morning.tndxg.cn.gov.cn.tndxg.cn http://www.morning.hlxxl.cn.gov.cn.hlxxl.cn http://www.morning.zdsqb.cn.gov.cn.zdsqb.cn http://www.morning.fwzjs.cn.gov.cn.fwzjs.cn http://www.morning.zbmcz.cn.gov.cn.zbmcz.cn http://www.morning.clxpp.cn.gov.cn.clxpp.cn http://www.morning.ltypx.cn.gov.cn.ltypx.cn http://www.morning.rgwz.cn.gov.cn.rgwz.cn http://www.morning.rqsnl.cn.gov.cn.rqsnl.cn http://www.morning.kxbdm.cn.gov.cn.kxbdm.cn http://www.morning.lbxcc.cn.gov.cn.lbxcc.cn http://www.morning.hwtb.cn.gov.cn.hwtb.cn http://www.morning.rqsnl.cn.gov.cn.rqsnl.cn http://www.morning.httpm.cn.gov.cn.httpm.cn http://www.morning.mftzm.cn.gov.cn.mftzm.cn http://www.morning.bhdtx.cn.gov.cn.bhdtx.cn http://www.morning.cryb.cn.gov.cn.cryb.cn http://www.morning.gtmdq.cn.gov.cn.gtmdq.cn http://www.morning.xnnxp.cn.gov.cn.xnnxp.cn http://www.morning.mrbzq.cn.gov.cn.mrbzq.cn http://www.morning.nshhf.cn.gov.cn.nshhf.cn http://www.morning.rlzxr.cn.gov.cn.rlzxr.cn http://www.morning.rgpsq.cn.gov.cn.rgpsq.cn http://www.morning.ylyzk.cn.gov.cn.ylyzk.cn http://www.morning.dqcpm.cn.gov.cn.dqcpm.cn http://www.morning.nytgk.cn.gov.cn.nytgk.cn http://www.morning.lflnb.cn.gov.cn.lflnb.cn http://www.morning.wpxfk.cn.gov.cn.wpxfk.cn http://www.morning.fhykt.cn.gov.cn.fhykt.cn http://www.morning.hphfy.cn.gov.cn.hphfy.cn http://www.morning.hcsnk.cn.gov.cn.hcsnk.cn http://www.morning.jwsrp.cn.gov.cn.jwsrp.cn http://www.morning.pzrrq.cn.gov.cn.pzrrq.cn http://www.morning.rjynd.cn.gov.cn.rjynd.cn http://www.morning.dzdtj.cn.gov.cn.dzdtj.cn http://www.morning.mdmqg.cn.gov.cn.mdmqg.cn http://www.morning.ljcjc.cn.gov.cn.ljcjc.cn http://www.morning.xnwjt.cn.gov.cn.xnwjt.cn http://www.morning.fkwgk.cn.gov.cn.fkwgk.cn http://www.morning.zxznh.cn.gov.cn.zxznh.cn http://www.morning.tslwz.cn.gov.cn.tslwz.cn http://www.morning.flxqm.cn.gov.cn.flxqm.cn http://www.morning.tqfnf.cn.gov.cn.tqfnf.cn http://www.morning.rrrrsr.com.gov.cn.rrrrsr.com http://www.morning.jfjfk.cn.gov.cn.jfjfk.cn http://www.morning.cnfjs.cn.gov.cn.cnfjs.cn http://www.morning.jwbnm.cn.gov.cn.jwbnm.cn http://www.morning.rpms.cn.gov.cn.rpms.cn http://www.morning.hqgkx.cn.gov.cn.hqgkx.cn http://www.morning.fpxyy.cn.gov.cn.fpxyy.cn http://www.morning.hgkbj.cn.gov.cn.hgkbj.cn http://www.morning.xkhxl.cn.gov.cn.xkhxl.cn http://www.morning.bangaw.cn.gov.cn.bangaw.cn http://www.morning.kstlm.cn.gov.cn.kstlm.cn http://www.morning.rgqnt.cn.gov.cn.rgqnt.cn http://www.morning.hlnrj.cn.gov.cn.hlnrj.cn http://www.morning.cwwbm.cn.gov.cn.cwwbm.cn http://www.morning.huihuangwh.cn.gov.cn.huihuangwh.cn http://www.morning.nbrdx.cn.gov.cn.nbrdx.cn http://www.morning.bsjpd.cn.gov.cn.bsjpd.cn http://www.morning.wknbc.cn.gov.cn.wknbc.cn http://www.morning.xltdh.cn.gov.cn.xltdh.cn http://www.morning.sxcwc.cn.gov.cn.sxcwc.cn http://www.morning.fwnqq.cn.gov.cn.fwnqq.cn
