重庆涪陵网站建设公司,东营做网站优化价格,全国网站集约化建设试点,网站放在服务器上大纲 Tumbling Count WindowsmapreduceWindow Size为2Window Size为3Window Size为4Window Size为5Window Size为6 完整代码参考资料 之前的案例中#xff0c;我们的Source都是确定内容的数据。而Flink是可以处理流式#xff08;Streaming#xff09;数据的#xff0c;就是… 大纲 Tumbling Count WindowsmapreduceWindow Size为2Window Size为3Window Size为4Window Size为5Window Size为6 完整代码参考资料 之前的案例中我们的Source都是确定内容的数据。而Flink是可以处理流式Streaming数据的就是数据会源源不断输入。 对于这种数据我们称之为无界流即没有“终止的界限”。但是程序在底层一定不能等着无止境的数据都传递结束再处理因为“无止境”就意味着“终止的界限”触发计算的条件是不存在的。那么我们可以人为的给它设置一个“界”这就是我们本节介绍的窗口。
Tumbling Count Windows
Tumbling Count Windows是指按元素个数计数的滚动窗口。 滚动窗口是指没有元素重叠的窗口比如下面图是个数为2的窗口。元素重叠的窗口我们会在《0基础学习PyFlink——个数滑动窗口Sliding Count Windows》介绍 个数为3的窗口 我们用代码探索下这个概念
map
word_count_data [(A,2),(A,1),(B,3),(B,1),(B,2),(C,3),(C,1),(C,4),(C,2),(D,3),(D,1),(D,4),(D,2),(D,5),(E,3),(E,1),(E,4),(E,2),(E,6),(E,5)]def word_count():env StreamExecutionEnvironment.get_execution_environment()env.set_runtime_mode(RuntimeExecutionMode.STREAMING)# write all the data to one fileenv.set_parallelism(1)source_type_info Types.TUPLE([Types.STRING(), Types.INT()])# define the source# mappgingsource env.from_collection(word_count_data, source_type_info)# source.print()# keyingkeyedsource.key_by(lambda i: i[0]) 这段代码构造了一个KeyedStream用于存储word_count_data中的数据。 我们并没有让Source是流的形式是因为为了降低例子复杂度。但是我们将runntime mode设置为流STREAMING模式。
reduce
我们需要定义一个Reduce类用于对元组中的数据进行计算。这个类需要继承于WindowFunction并实现相应方法本例中是apply。 apply会计算一个相同key的元素个数。比如key是“E”的元组个数是6。
class SumWindowFunction(WindowFunction[tuple, tuple, str, CountWindow]):def apply(self, key: str, window: CountWindow, inputs: Iterable[tuple]):return [(key, len([e for e in inputs]))]Window Size为2 # reducingreducedkeyed.count_window(2) \.apply(SumWindowFunction(),Types.TUPLE([Types.STRING(), Types.INT()]))# # define the sinkreduced.print()# submit for executionenv.execute()(A,2) (B,2) (C,2) (C,2) (D,2) (D,2) (E,2) (E,2) (E,2) A的个数是2是因为A的确只有两个元组而一个Size为2的Window正好承载了这两个元素。于是有A,2这个结果B的个数是3。但是会产生两个窗口第一个窗口承载了前两个元素第二个窗口当前只有一个元素。于是第一个窗口进行了Reduce计算得出一个(B,2)第二个窗口还没进行reduce计算就没有展现出结果C有4个正好可以被2个窗口承载。这样我们就看到2个(C,2)。D有5个情况和B类似。它被分成了3个窗口只有2个窗口满足个数条件于是就输出2个(D,2)最后一个窗口因为元素不够就没尽兴reduce计算了。E有6个正好被3个窗口承载。我们就看到3个(E,2)。
Window Size为3 # reducingreducedkeyed.count_window(3) \.apply(SumWindowFunction(),Types.TUPLE([Types.STRING(), Types.INT()]))(B,3) (C,3) (D,3) (E,3) (E,3) Window Size为4 # reducingreducedkeyed.count_window(4) \.apply(SumWindowFunction(),Types.TUPLE([Types.STRING(), Types.INT()]))(C,4) (D,4) (E,4) Window Size为5 # reducingreducedkeyed.count_window(5) \.apply(SumWindowFunction(),Types.TUPLE([Types.STRING(), Types.INT()]))(D,5) (E,5) Window Size为6 # reducingreducedkeyed.count_window(6) \.apply(SumWindowFunction(),Types.TUPLE([Types.STRING(), Types.INT()]))(E,6) 完整代码
from typing import Iterablefrom pyflink.common import Types
from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode, WindowFunction
from pyflink.datastream.window import CountWindowclass SumWindowFunction(WindowFunction[tuple, tuple, str, CountWindow]):def apply(self, key: str, window: CountWindow, inputs: Iterable[tuple]):return [(key, len([e for e in inputs]))]word_count_data [(A,2),(A,1),(B,3),(B,1),(B,2),(C,3),(C,1),(C,4),(C,2),(D,3),(D,1),(D,4),(D,2),(D,5),(E,3),(E,1),(E,4),(E,2),(E,6),(E,5)]def word_count():env StreamExecutionEnvironment.get_execution_environment()env.set_runtime_mode(RuntimeExecutionMode.STREAMING)# write all the data to one fileenv.set_parallelism(1)source_type_info Types.TUPLE([Types.STRING(), Types.INT()])# define the source# mappgingsource env.from_collection(word_count_data, source_type_info)# source.print()# keyingkeyedsource.key_by(lambda i: i[0]) # reducingreducedkeyed.count_window(2) \.apply(SumWindowFunction(),Types.TUPLE([Types.STRING(), Types.INT()]))# # define the sinkreduced.print()# submit for executionenv.execute()if __name__ __main__:word_count()参考资料
https://nightlies.apache.org/flink/flink-docs-release-1.18/zh/docs/learn-flink/streaming_analytics/
