Shulou(Shulou.com)06/01 Report--

This article mainly explains "what is the role of Time and Window in Flink". The content of the explanation is simple and clear, and it is easy to learn and understand. Please follow the editor's train of thought to study and learn "what is the role of Time and Window in Flink"?

Preface

Flink is a streaming, real-time computing engine

There are two concepts in the above sentence, one is streaming, the other is real-time.

Streaming: that is, the data keeps flowing in, that is, the data has no boundary, but when we calculate it, it must be carried out within a boundary, so there is a problem, how to determine the boundary? There are no more than two ways to determine according to the time period or the amount of data, a boundary is divided according to how often the time period is, and a boundary is divided according to the amount of data. This is how the boundary is divided in Flink, which will be explained in detail in this article.

Real-time: that is, after the data is sent, the relevant calculations are carried out immediately, and then the results are output. There are two kinds of calculations here:

One is to calculate only the data within the boundary, which is easy to understand, such as counting the number of news browsed by each user in the last five minutes, you can take all the data in the last five minutes, and then group according to each user to count the total number of news.

The other is to associate intra-boundary data with external data, for example, to count the regions from which users browsing news in the last five minutes come from. This requires associating the information of users browsing news within five minutes with the regional dimension table in hive, and then doing relevant calculations.

Time and WindowTime

In Flink, time is an extremely important field if the boundary is divided by time period.

There are three types of time in Flink, as shown in the following figure:

Event Time: is the time when the event was created. It is usually described by the timestamp in the event, for example, in the collected log data, each log records its own generation time, and Flink accesses the event timestamp through the timestamp allocator.

Ingestion Time: is the time when the data enters the Flink.

Processing Time: is the local system time for each operator that performs time-based operations, machine-dependent, and the default time attribute is Processing Time.

For example, the time for a log to enter Flink is 2021-01-22, and the system time for Window is 2021-01-22. The contents of the log are as follows:

2021-01-06 18 3715 624 INFO Fail over to rm2

For the business, which time is the most meaningful to count the number of fault logs in the 1min? -- eventTime, because we need to make statistics according to the time when the log is generated.

Window

Window, or window, the boundary we've been talking about is the Window here.

Official explanation: streaming computing is a data processing engine designed to handle infinite data sets, which refer to a growing set of essentially infinite data sets, while window is a means of cutting infinite data into finite blocks for processing.

So Window is the core of infinite data flow processing. Window splits an infinite stream into finite size "buckets" buckets on which we can do calculations.

Window Typ

As mentioned at the beginning of this article, there are two ways to divide windows:

Intercept according to time (time-driven-window), such as every 1 minute or every 10 minutes.

Data-driven-window based on data, such as every 5 data or 50 data.

Window type

For TimeWindow (dividing windows according to time), they can be divided into three categories according to the principle of window implementation: scrolling window (Tumbling Window), sliding window (Sliding Window) and session window (Session Window).

Scroll window (Tumbling Windows)

Slice the data according to a fixed window length.

Features: time alignment, fixed window length, no overlap.

The scrolling window allocator assigns each element to a window of a specified window size, which has a fixed size and does not overlap.

For example, if you specify a 5-minute scrolling window, the window is created as shown in the following figure:

Scroll window

Applicable scenarios: suitable for BI statistics (aggregate calculation for each time period).

Sliding window (Sliding Windows)

Sliding window is a more generalized form of fixed window, which consists of fixed window length and sliding interval.

Features: time alignment, fixed window length, overlap.

The sliding window allocator assigns elements to fixed-length windows, similar to scrolling windows, the size of the window is configured by the window size parameter, and another window sliding parameter controls the frequency at which the sliding window starts. Therefore, if the sliding parameter of the sliding window is less than the window size, the window can overlap, in which case the element will be assigned to multiple windows.

For example, if you have a 10-minute window and a 5-minute slide, the 5-minute window in each window contains the data generated in the last 10 minutes, as shown in the following figure:

Sliding window

Applicable scenario: statistics in the most recent time period (calculate the failure rate of the recent 5min of an interface to decide whether to report to the police).

Session window (Session Windows)

It consists of a series of events combined with a timeout gap of a specified length of time, similar to the session of web applications, that is, a new window will be generated if no new data is received for a period of time.

Features: time is not aligned.

The session window allocator groups elements through session activities. Compared with scrolling windows and sliding windows, session windows do not overlap and have fixed start and end times. On the contrary, when it no longer receives elements within a fixed period of time, that is, inactive intervals are generated, the window closes. A session window is configured with a session interval that defines the length of the inactive period. When this inactive cycle occurs, the current session will be closed and subsequent elements will be assigned to the new session window.

Session window Window APITimeWindow

TimeWindow is to compose all the data within a specified time range into a window, and calculate all the data in a window at one time (that is, the data within a boundary is calculated at the beginning of this article).

Let's take the number of cars passing at traffic lights as an example:

There will be cars passing at the traffic lights, and it is impossible to calculate how many cars will pass. Because of the continuous flow of traffic, the calculation has no boundary.

So we count the number of cars passing red street lights every 15 seconds, such as 2 cars in the first 15 seconds, 3 cars in the second 15 seconds, and 1 car in the third 15 seconds.

Tumbling-time-window (no overlapping data)

We use the nc command in Linux to simulate the sender of the data

11. Open the sending port with a port number of 9999

2nc-lk 9999

three

forty-two。 Send content (key represents different intersection, value represents vehicles passing each time)

5 send one line at a time, the sending interval represents the time interval of the car passing by

69,3

79,2

89,7

94,9

102,6

111,5

122,3

135,7

145,4

Flink collects data and calculates:

1object Window {

2 def main (args: Array [String]): Unit = {

3 / / TODO time-window

4 / / 1. Create a running environment

5 val env = StreamExecutionEnvironment.getExecutionEnvironment

six

7 / / 2. Define the source of data flow

8 val text = env.socketTextStream ("localhost", 9999)

nine

10 / / 3. Convert data format, text- > CarWc

11 case class CarWc (sensorId: Int, carCnt: Int)

12 val ds1: DataStream [CarWc] = text.map {

13 line = > {

14 val tokens = line.split (",")

15 CarWc (tokens (0) .trim.toInt, tokens (1) .trim.toInt)

16}

17}

eighteen

19 / / 4. Perform statistical operations, one tumbling window per sensorId, and the window size is 5 seconds.

20 / / that is, every 5 seconds, the number of cars passing through traffic lights at each intersection in the past 5 seconds.

21 val ds2: DataStream [CarWc] = ds1

22. Keyby ("sensorId")

23. TimeWindow (Time.seconds (5))

24. Sum ("carCnt")

twenty-five

26 / / 5. Show statistical results

27 ds2.print ()

twenty-eight

29 / / 6. Trigger flow calculation

30 env.execute (this.getClass.getName)

thirty-one

32}

33}

The data we send does not specify a time field, so Flink uses the default Processing Time, which is the time when the Flink system processes the data.

Sliding-time-window (with overlapping data)

1//1. Create a running environment

2val env = StreamExecutionEnvironment.getExecutionEnvironment

three

4//2. Define the source of data flow

5val text = env.socketTextStream ("localhost", 9999)

six

7//3. Convert data format, text- > CarWc

8case class CarWc (sensorId: Int, carCnt: Int)

nine

10val ds1: DataStream [CarWc] = text.map {

11 line = > {

12 val tokens = line.split (",")

13 CarWc (tokens (0) .trim.toInt, tokens (1) .trim.toInt)

14}

15}

16//4. Perform statistical operations, one sliding window per sensorId, window time 10 seconds, sliding time 5 seconds

In other words, every 5 seconds, count the number of cars passing through traffic lights at each intersection in the past 10 seconds.

18val ds2: DataStream [CarWc] = ds1

19. Keyby ("sensorId")

20. Timewindow (Time.seconds (10), Time.seconds (5))

21. Sum ("carCnt")

twenty-two

23//5. Show statistical results

24ds2.print ()

twenty-five

26//6. Trigger flow calculation

27env.execute (this.getClass.getName) CountWindow

CountWindow triggers execution based on the number of the same key elements in the window, and only calculates the result of the key whose number of elements reaches the size of the window.

Note: the window_size of CountWindow refers to the number of elements of the same Key, not the total number of elements entered.

Tumbling-count-window (no overlapping data)

1//1. Create a running environment

2val env = StreamExecutionEnvironment.getExecutionEnvironment

three

4//2. Define the source of data flow

5val text = env.socketTextStream ("localhost", 9999)

six

7//3. Convert data format, text- > CarWc

8case class CarWc (sensorId: Int, carCnt: Int)

nine

10val ds1: DataStream [CarWc] = text.map {

11 (f) = > {

12 val tokens = f.split (",")

13 CarWc (tokens (0) .trim.toInt, tokens (1) .trim.toInt)

14}

15}

16//4. Perform statistical operations, one tumbling window per sensorId, with a window size of 5

17Universe / collect according to key, the corresponding key appears 5 times as a result

18val ds2: DataStream [CarWc] = ds1

19. Keyby ("sensorId")

20. CountWindow (5)

21. Sum ("carCnt")

twenty-two

23//5. Show statistical results

24ds2.print ()

twenty-five

26//6. Trigger flow calculation

27env.execute (this.getClass.getName)

Sliding-count-window (with overlapping data)

It is also the operation of window length and sliding window: the window length is 5 and the sliding length is 3.

1//1. Create a running environment

2val env = StreamExecutionEnvironment.getExecutionEnvironment

three

4//2. Define the source of data flow

5val text = env.socketTextStream ("localhost", 9999)

six

7//3. Convert data format, text- > CarWc

8case class CarWc (sensorId: Int, carCnt: Int)

nine

10val ds1: DataStream [CarWc] = text.map {

11 (f) = > {

12 val tokens = f.split (",")

13 CarWc (tokens (0) .trim.toInt, tokens (1) .trim.toInt)

14}

15}

16//4. Perform statistical operations. There is one sliding window for each sensorId. The size of the window is 3 pieces of data, and the window slides to 3 pieces of data.

In other words, statistics are made at each intersection. When you receive three messages about it, the number of cars passing through each intersection is counted in the last five messages.

18val ds2: DataStream [CarWc] = ds1

19. Keyby ("sensorId")

20. CountWindow (5,3)

21. Sum ("carCnt")

twenty-two

23//5. Show statistical results

24ds2.print ()

twenty-five

26//6. Trigger flow calculation

27env.execute (this.getClass.getName)

Window summary

Flink supports two ways of dividing windows (time and count)

If the window is divided according to time, then it is a time-window

If you divide the window according to the data, then it is a count-window

Flink supports two important properties of the window (size and interval)

If size=interval, then tumbling-window will be formed (no overlapping data)

If size > interval, then sliding-window (with overlapping data) is formed.

If size (item, 1) .keyby (0)

fifteen

16 / / introduce the time window

17 val streamWindow = streamKeyBy.timeWindow (Time.seconds (5))

eighteen

19 / / perform aggregation operation

20 val streamReduce = streamWindow.reduce (

21 (item1, item2) = > (item1._1, item1._2 + item2._2)

22)

twenty-three

24 / / write aggregated data to a file

25 streamReduce.print ()

twenty-six

27 / / Executive procedure

28 env.execute ("TumblingWindow")

29}

30} Window Apply

The apply method can be customized through anonymous inner class methods. Used when there are some complex calculations.

Usage

Implement a WindowFunction class

Specify the generic type of this class as [input data type, output data type, type of grouped field used in keyBy, window type]

Example: use the apply method to implement word statistics

Steps:

Get the stream processing runtime environment

Build the socket stream data source and specify the IP address and port number

Docking the received data into word tuples

Use keyBy for shunt (grouping)

Use timeWinodw to specify the length of the window (calculated every 3 seconds)

Implement an WindowFunction anonymous inner class

Implementation of aggregate Computing in apply method

Using Collector.collect to collect data

The core code is as follows:

1 / / 1. Get the stream processing runtime environment

2 val env = StreamExecutionEnvironment.getExecutionEnvironment

three

4 / / 2. Build the socket stream data source and specify the IP address and port number

5 val textDataStream = env.socketTextStream ("node01", 9999) .flatMap (_ .split (""))

six

7 / / 3. Docking the received data into word tuples

8 val wordDataStream = textDataStream.map (_-> 1)

nine

10 / / 4. Use keyBy for shunt (grouping)

11 val groupedDataStream: KeyedStream [(String, Int), String] = wordDataStream.keyBy (_. 1)

twelve

13 / / 5. Use timeWinodw to specify the length of the window (calculated every 3 seconds)

14 val windowDataStream: WindowedStream [(String, Int), String, TimeWindow] = groupedDataStream.timeWindow (Time.seconds (3))

fifteen

16 / / 6. Implement an WindowFunction anonymous inner class

17 val reduceDatStream: DataStream [(String, Int)] = windowDataStream.apply (new RichWindowFunction [(String, Int), (String, Int), String, TimeWindow] {

18 / / data aggregation in apply method

19 override def apply (key: String, window: TimeWindow, input: Iterable [(String, Int)], out: Collector [(String, Int)]): Unit = {

20 println ("hello world")

21 val tuple = input.reduce ((T1, T2) = > {

22 (t1.room1, t1.room2 + t2.room2)

23})

24 / / the data to be returned is collected and sent back

25 out.collect (tuple)

26}

27})

28 reduceDatStream.print ()

29 env.execute () Window Fold

WindowedStream → DataStream: a function that assigns a fold function to a window and returns a result after fold.

1import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

2import org.apache.flink.api.scala._

3import org.apache.flink.streaming.api.windowing.time.Time

four

5object StreamWindowFold {

6 def main (args: Array [String]): Unit = {

7 / / get the execution environment

8 val env = StreamExecutionEnvironment.getExecutionEnvironment

nine

10 / / create a SocketSource

11 val stream = env.socketTextStream ("node01", 9999pr.

twelve

13 / / A pair of stream is processed and aggregated by key

14 val streamKeyBy = stream.map (item = > (item, 1)) .keyby (0)

fifteen

16 / / introduce a scrolling window

17 val streamWindow = streamKeyBy.timeWindow (Time.seconds (5))

eighteen

19 / / perform fold operation

20 val streamFold = streamWindow.fold (100) {

21 (begin, item) = >

22 begin + item._2

23}

twenty-four

25 / / write aggregated data to a file

26 streamFold.print ()

twenty-seven

28 / / execute the program

29 env.execute ("TumblingWindow")

30}

31} Aggregation on Window

WindowedStream → DataStream: aggregates all elements within a window. The difference between min and minBy is that min returns a minimum value, while minBy returns an element that contains a minimum value field (the same principle applies to max and maxBy).

1import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

2import org.apache.flink.streaming.api.windowing.time.Time

3import org.apache.flink.api.scala._

four

5object StreamWindowAggregation {

6 def main (args: Array [String]): Unit = {

7 / / get the execution environment

8 val env = StreamExecutionEnvironment.getExecutionEnvironment

nine

10 / / create a SocketSource

11 val stream = env.socketTextStream ("node01", 9999)

twelve

13 / / A pair of stream is processed and aggregated by key

14 val streamKeyBy = stream.map (item = > (item.split (") (0), item.split (") (1)) .keyby (0)

fifteen

16 / / introduce a scrolling window

17 val streamWindow = streamKeyBy.timeWindow (Time.seconds (5))

eighteen

19 / / perform aggregation operation

20 val streamMax = streamWindow.max (1)

twenty-one

22 / / write aggregated data to a file

23 streamMax.print ()

twenty-four

25 / / execute the program

26 env.execute ("TumblingWindow")

27}

28} the introduction of EventTime and WindowEventTime

It is inconsistent with the time in the real world. In flink, it is divided into three types: event time, extraction time and processing time.

If the time window is defined on the basis of EventTime, it will form an EventTimeWindow, which requires that the message itself should carry EventTime.

If the time window is defined on the basis of IngesingtTime, it will form an IngestingTimeWindow, based on the systemTime of source.

If the time window is defined based on the ProcessingTime benchmark, it will form a ProcessingTimeWindow, based on the systemTime of operator.

In the streaming processing of Flink, the vast majority of businesses will use eventTime, generally only when eventTime can not be used, will be forced to use ProcessingTime or IngestionTime.

If you want to use EventTime, you need to introduce the time attribute of EventTime, as shown below:

1val env = StreamExecutionEnvironment.getExecutionEnvironment

two

3max / append time characteristics to each stream created by env from the time of the call

Introduction of 4env.setStreamTimeCharacteristic (TimeCharacteristic.EventTime) Watermark

We know that there is a process and time in stream processing from the generation of events to the flow through source and then to operator. Although in most cases, the data flowing to operator comes according to the time order of events, it does not rule out the occurrence of disorder due to network, back pressure and other reasons. The so-called disorder means that the sequence of events received by Flink is not strictly arranged according to the Event Time order of events. So when Flink was originally designed, it took into account the network delay, network disorder and other problems, so put forward an abstract concept: watermark (WaterMark)

As shown in the figure above, there is a problem. Once there is disorder, if we only decide the operation of Window based on EventTime, we cannot determine whether all the data are in place, but we cannot wait indefinitely. At this time, we must have a mechanism to ensure that after a specific time, we must trigger Window to calculate. This special mechanism is Watermark.

Watermark is used to deal with out-of-order events, while the correct handling of out-of-order events is usually implemented by Watermark mechanism combined with Window.

The Watermark in the data stream is used to represent data whose timestamp is less than Watermark and has arrived, so the execution of Window is also triggered by Watermark.

Watermark can be understood as a delay trigger mechanism. We can set the delay time t of Watermark. Each time the system verifies the largest maxEventTime in the data that has arrived, and then determines that all data whose EventTime is less than maxEventTime-t has arrived. If the stop time of a window is equal to maxEventTime-t, then the window is triggered to execute.

The Watermarker of ordered flow is shown in the following figure: (Watermark is set to 0)

Watermark of ordered data

The Watermarker of the disordered flow is shown in the following figure: (Watermark is set to 2)

Watermark of unordered data

When the Flink receives each piece of data, it generates a Watermark, which Watermark is equal to the maxEventTime-delay time in all the current arrival data, that is, the Watermark is carried by the data. Once the Watermark carried by the data is later than the stop time of the current untriggered window, the execution of the corresponding window will be triggered. Because Watermark is carried by data, if new data cannot be obtained during the run, windows that are not triggered will never be triggered.

In the figure above, we set a maximum delay arrival time of 2s, so the Watermark for an event with a timestamp of 7s is 5s, and the Watermark for an event with a timestamp of 12s is 10s. If our window 1 is 1s~5s and window 2 is 6s~10s, then the Watermarker of the event with a timestamp of 7s just triggers window 1, and the Watermark with the timestamp of 12s triggers window 2.

The processing of late data by Flink

WaterMark and Window mechanisms solve the problem of streaming data out of order. For data that is out of order due to delay, business processing can be carried out according to eventTime, and Flink for delayed data also has its own solution. The main method is to give a time allowed to delay, and it is still acceptable to process delayed data within this time range.

The time allowed for delay is set through allowedLateness (lateness: Time).

The delay data is saved through sideOutputLateData (outputTag: OutputTag [T]).

Delay data is obtained through DataStream.getSideOutput (tag: OutputTag [X]).

The specific usage is as follows:

AllowedLateness (lateness: Time) 1def allowedLateness (lateness: Time): WindowedStream [T, K, W] = {

2 javaStream.allowedLateness (lateness)

3 this

4}

This method passes in a time value that sets the time at which data is allowed to be late, which is different from the concept of time in WaterMark. Let's review it again:

Event time for WaterMark= data-disordered time values are allowed

As the new data arrives, the value of waterMark is updated to the latest data event time-the out-of-order time value is allowed, but if a piece of historical data comes at this time, the watermark value is not updated. Generally speaking, waterMark is designed to receive as much out-of-order data as possible.

Well, the time value here is mainly for waiting for late data. Within a certain time range, if the data belonging to this window arrives, it will still be calculated. The calculation method will be explained in detail later.

Note: this method is only for event-time-based windows, and if it is processing-time-based and specifies a non-zero time value, an exception will be thrown.

SideOutputLateData (outputTag: OutputTag [T]) 1def sideOutputLateData (outputTag: OutputTag [T]): WindowedStream [T, K, W] = {

2 javaStream.sideOutputLateData (outputTag)

3 this

4}

This method saves belated data to a given outputTag parameter, and OutputTag is an object used to mark delayed data.

DataStream.getSideOutput (tag: OutputTag [X])

This method is called by the DataStream returned by operations such as window, and the object marked with delay data is passed in to get the delayed data.

Understanding of delayed data

Delayed data refers to:

After the current window [assuming the window range is 10-15] has been calculated, there is another data belonging to the window [assuming the event time is 13]. At this time, the Window operation will still be triggered, which is called deferred data.

So the question is, how to calculate the delay time?

Assuming that the window range is 10-15 and the delay time is 2s, as long as the WaterMark=15+2,10-15 window is used, the Window operation can no longer be triggered, even if the Event Time of the new data belongs to this window time.

Flink Associated Hive Partition Table

Flink 1.12 supports the function of the latest partition of Hive as a temporal table, which can be directly associated with the latest partition of the Hive partition table through SQL, and will automatically listen to the latest Hive partition, and automatically replace all the dimension table data when the new partition is monitored. In this way, users can complete the real-time association of Kafka streams with the latest Hive partitions to achieve data broadening without writing DataStream programs.

Specific usage:

Register HiveCatalog with Sql Client:

1vim conf/sql-client-defaults.yaml

2catalogs:

3-name: hive_catalog

4 type: hive

5 hive-conf-dir: / disk0/soft/hive-conf/ # this directory requires a package hive-site.xml file

Create a Kafka table

1CREATE TABLE hive_catalog.flink_db.kfk_fact_bill_master_12 (

2 master Row

3proctime as PROCTIME ()-- PROCTIME is used to associate with the Hive temporal table

4) WITH (

5 'connector' =' kafka'

6 'topic' =' topic_name'

7 'format' =' json'

8 'properties.bootstrap.servers' =' host:9092'

9 'properties.group.id' =' flinkTestGroup'

10 'scan.startup.mode' =' timestamp'

11 'scan.startup.timestamp-millis' =' 1607844694000'

12)

The Flink fact table is associated with the latest Hive partition data

Dim_extend_shop_info is an existing table in Hive, so we use table hint to turn on dimension table parameters dynamically.

1CREATE VIEW IF NOT EXISTS hive_catalog.flink_db.view_fact_bill_master as

2SELECT * FROM

3 (select t. 1, t2.group_id, t2.shop_id, t2.group_name, t2.shop_name, t2.brand_id

4 ROW_NUMBER () OVER (PARTITION BY groupID, shopID, orderKey ORDER BY actionTime desc) rn

5 from hive_catalog.flink_db.kfk_fact_bill_master_12 t1

6 JOIN hive_catalog.flink_db.dim_extend_shop_info

7 / * + OPTIONS ('streaming-source.enable'='true'

8 'streaming-source.partition.include' =' latest'

9 'streaming-source.monitor-interval' =' 1h'

10 'streaming-source.partition-order' =' partition-name') * /

11 FOR SYSTEM_TIME AS OF t1.proctime AS T2-temporal table

12 ON t1.groupID = t2.group_id and t1.shopID = t2.shop_id

13 where groupID in (202042) t where t.rn = 1

Parameter explanation:

Streaming-source.enable enables streaming reading of Hive data.

There are two values for streaming-source.partition.include:

Latest attribute: only the latest partition data is read.

All: read all partition data. The default value is all, which means to read all partitions. Latest can only be used in temporal join to read the latest partition as a dimension table, not the latest partition data directly.

The time for streaming-source.monitor-interval to listen to the generation of a new partition should not be too short, the shortest is 1 hour, because the current implementation is that every task will query metastore, and high-frequency checking may exert too much pressure on metastore. It should be noted that 1.12.1 loosens this restriction, but it is still recommended not to have a too short interval according to the actual business.

There are three main streaming-source.partition-order partitioning policies, of which partition-name is the most recommended:

Partition-name loads the latest partition using the default partition name order

Create-time uses partition files to create a chronological order

Partition-time uses partition time order

Thank you for reading, the above is the content of "what is the role of Time and Window in Flink". After the study of this article, I believe you have a deeper understanding of the role of Time and Window in Flink, and the specific use needs to be verified in practice. Here is, the editor will push for you more related knowledge points of the article, welcome to follow!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.