Flink多并行度与WaterMark

最近在复习flink的时候，发现之前写的demo全是单并行度，突发奇想多并行度下的窗口触发是否与单并行度相同? 故而就引申出了下列一些列问题。
首先我设置了数据延迟时间为2s，然后设置了一个长度为5秒滑动距离为1s的窗口。

测试数据

a 1
b 2
c 3
a 4
f 5
z 6
v 7

其中数字部分为时间戳

单并行度

当a 4 进入后触发【-3~2】的窗口计算（窗口计算如下）

窗口起始时间计算逻辑

// org.apache.flink.streaming.api.windowing.windows.TimeWindow 函数位置

    public static long getWindowStartWithOffset(long timestamp, long offset, long windowSize) {
        return timestamp - (timestamp - offset + windowSize) % windowSize;
    }

根据如上规则得出如下公式,

1 - （1 - 2 + 5）% 5 = -3

# 1 事件时间
# 2 延时时间
# 5 窗口长度

多并行度

当把并行度设置为4的时候 “诡异”的时间就产生了数据发送到 a 4的时候没有触发任何计算于是我继续发送数据

f 5

z 6

v 7

至此才触发【-3~2】的窗口计算。令我百思不得其解！深究下去，得出如下结论。

所有分区数据都要满足【当前分区的事件时间 >= 所有分区当前最小的watermark】如下

上图所示第一个窗口应该是【-3~2】,所有分区中最小的watermark应该为4。但当时间时间4的数据过来后窗口并未关闭，原因是每个分区都维护了一份自己的waterMark。

此时只有p4的时间达到了触发窗口【-3_{2】关闭的标准。所以【-3}2】窗口并不会关闭
此时p1 进入一条事件时间为5的数据也达到了触发窗口【-3~2】关闭的标准。但p2、p3依旧没有达到触发时间的标准，所以窗口仍旧没有关闭

截止到事件时间7进入p3为止，所有分区数据均已超过最小watermark4，触发窗口【-3~2】关闭

引申：假设6进入p2后，乱序数据小于4，比如是3进入到p3,后续频繁出现这种情况。导致窗口【-3~2】长时间无法关闭，数据全部积压如何解决？

附上测试代码

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
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.SlidingEventTimeWindows;
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.time.Duration;


public class Test {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
       // executionEnvironment.setParallelism(4);

        DataStreamSource source = executionEnvironment.socketTextStream("node28.testbigdate", 9991);

        SingleOutputStreamOperator map = source.map(line -> {
            JSONObject json = new JSONObject();
            String[] s = line.split(" ");
            json.put("name", s[0]);
            json.put("date", s[1]);
            return json;
        }).assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("date") * 1000L;
                    }
                }));
        map.print("===========>");

        KeyedStream name = map.keyBy(t -> t.getString("name"));
        name.window(SlidingEventTimeWindows.of(Time.seconds(5), Time.seconds(1)))
                .process(new ProcessWindowFunction() {

                    @Override
                    public void process(String s, Context context, Iterable elements, Collector

Flink多并行度与WaterMark

大数据系统相关栏目本月热门文章