会话窗口 flink

Question

有人可以帮我了解flink中的窗口（会话）何时以及如何发生吗？或者样品是如何处理的？

例如，如果我有连续的事件流流入，事件是应用程序中的请求和应用程序提供的响应。 作为 flink 处理的一部分，我们需要了解处理请求需要多少时间。

我知道有时间翻滚窗口每 n 秒触发一次，这是配置的，一旦时间流逝，该时间窗口中的所有事件将被聚合。

例如： 假设定义的时间窗口是 30 秒，如果一个事件在 t 时间到达，另一个在 t+30 到达，那么两者都将被处理，但在 t+31 到达的事件将被忽略。

如果我说的上述说法不正确，请更正。

上面的问题是：如果说一个事件在t时间到达，另一个事件在t+3时间到达，是否还要等待整整30秒才能汇总并最终确定结果？

现在在会话窗口的情况下，这是如何工作的？如果事件是单独处理的，并且在反序列化时将代理时间戳用作单个事件的 session_id，那么将为每个事件创建会话窗口吗？如果是，那么我们是否需要区别对待请求和响应事件，因为如果我们不这样做，那么响应事件不会有自己的会话窗口吗？

我将尝试在短时间内发布我正在使用的示例（在 java 中），但以上几点的任何输入都会有所帮助！

处理函数

DTO：

public class IncomingEvent{
    private String id;
    private String eventId;
    private Date timestamp;
    private String component;
    //getters and setters
}
public class FinalOutPutEvent{
    private String id;
    private long timeTaken;
    //getters and setters
}

================================================ 传入事件的反序列​​化：

公共类 IncomingEventDeserializationScheme 实现 KafkaDeserializationSchema {

private ObjectMapper mapper;

public IncomingEventDeserializationScheme(ObjectMapper mapper) {
    this.mapper = mapper;
}

@Override
public TypeInformation<IncomingEvent> getProducedType() {
    return TypeInformation.of(IncomingEvent.class);
}

@Override
public boolean isEndOfStream(IncomingEvent nextElement) {
    return false;
}

@Override
public IncomingEvent deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
    if (record.value() == null) {
        return null;
    }
    try {
        IncomingEvent event = mapper.readValue(record.value(), IncomingEvent.class);
        if(event != null) {
            new SessionWindow(record.timestamp());
            event.setOffset(record.offset());
            event.setTopic(record.topic());
            event.setPartition(record.partition());
            event.setBrokerTimestamp(record.timestamp());
        }
        return event;
    } catch (Exception e) {
        return null;
    }
}

}

================================================

主要逻辑

     public class MyEventJob {

private static final ObjectMapper mapper = new ObjectMapper();

public static void main(String[] args) throws Exception {

    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    MyEventJob eventJob = new MyEventJob();

    InputStream inStream = eventJob.getFileFromResources("myConfig.properties");
    ParameterTool parameter = ParameterTool.fromPropertiesFile(inStream);
    Properties properties = parameter.getProperties();
    Integer timePeriodBetweenEvents = 120;
    String outWardTopicHostedOnServer = localhost:9092";
    DataStreamSource<IncomingEvent> stream = env.addSource(new FlinkKafkaConsumer<>("my-input-topic", new IncomingEventDeserializationScheme(mapper), properties));
    SingleOutputStreamOperator<IncomingEvent> filteredStream = stream
        .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<IncomingEvent>() {
            long eventTime;
            @Override
            public long extractTimestamp(IncomingEvent element, long previousElementTimestamp) {
                return element.getTimestamp();
            }
            @Override
            public Watermark getCurrentWatermark() {
                return new Watermark(eventTime); 
            }
        })
        .map(e -> { e.setId(e.getEventId()); return e; });
    SingleOutputStreamOperator<FinalOutPutEvent> correlatedStream = filteredStream
        .keyBy(new KeySelector<IncomingEvent, String> (){
            @Override
            public String getKey(@Nonnull IncomingEvent input) throws Exception {
                return input.getId();
            }
        })
        .window(GlobalWindows.create()).allowedLateness(Time.seconds(defaultSliceTimePeriod))
        .trigger( new Trigger<IncomingEvent, Window> (){
            private final long sessionTimeOut;
            public SessionTrigger(long sessionTimeOut) {
                this.sessionTimeOut = sessionTimeOut;
            }
            @Override
            public TriggerResult onElement(IncomingEvent element, long timestamp, Window window, TriggerContext ctx)
                    throws Exception {
                ctx.registerProcessingTimeTimer(timestamp + sessionTimeOut); 
                return TriggerResult.CONTINUE;
            }
            @Override
            public TriggerResult onProcessingTime(long time, Window window, TriggerContext ctx) throws Exception {
                return TriggerResult.FIRE_AND_PURGE;
            }
            @Override
            public TriggerResult onEventTime(long time, Window window, TriggerContext ctx) throws Exception {
                    return TriggerResult.CONTINUE;
            }
            @Override
            public void clear(Window window, TriggerContext ctx) throws Exception {
                //check the clear method implementation
            }
        })
        .process(new ProcessWindowFunction<IncomingEvent, FinalOutPutEvent, String, SessionWindow>() {
        @Override
        public void process(String arg0,
                ProcessWindowFunction<IncomingEvent, FinalOutPutEvent, String, SessionWindow>.Context arg1,
                Iterable<IncomingEvent> input, Collector<FinalOutPutEvent> out) throws Exception {
            List<IncomingEvent> eventsIn = new ArrayList<>();
            input.forEach(eventsIn::add);
            if(eventsIn.size() == 1) {
                //Logic to handle incomplete request/response events
            } else if (eventsIn.size() == 2) {
                //Logic to handle the complete request/response and how much time it took
            }
        }
    } );
        FlinkKafkaProducer<FinalOutPutEvent> kafkaProducer = new FlinkKafkaProducer<>(
                outWardTopicHostedOnServer,            // broker list
                "target-topic",            // target topic
                new EventSerializationScheme(mapper));
    correlatedStream.addSink(kafkaProducer);
    env.execute("Streaming");
}

}

谢谢 维姬

Answer 1

根据您的描述，我认为您想编写一个自定义的ProcessFunction，由session_id 键入。您将拥有一个ValueState，用于存储请求事件的时间戳。当您获得相应的响应事件时，您计算增量并发出（使用session_id）并清除状态。

您可能还想在收到请求事件时设置一个计时器，这样如果您在安全/长时间内没有收到响应事件，您可以发出失败请求的侧面输出。

Answer 2

因此，使用默认触发器，每个窗口都会在时间完全过去后完成。取决于您使用的是EventTime 还是ProcessingTime，这可能意味着不同的事情，但一般来说，Flink 将始终等待窗口关闭，然后再完全处理。在您的情况下，t+31 的事件将简单地转到另一个窗口。

对于会话窗口，它们也是窗口，这意味着最终它们只是聚合时间戳之间的差异低于定义的间隙的样本。在内部，这比普通窗口更复杂，因为它们没有定义开始和结束。会话窗口操作员获取样本并为每个单独的样本创建一个新窗口。然后，操作员验证新创建的窗口是否可以与已经存在的窗口合并（即它们的时间戳是否比间隙更近）并合并它们。这最终导致窗口的所有元素的时间戳比定义的间隙更接近。

Answer 3

你让这变得比它需要的更复杂。下面的示例需要进行一些调整，但希望能够传达如何使用KeyedProcessFunction 而不是会话窗口的想法。

另外，BoundedOutOfOrdernessTimestampExtractor 的构造函数需要传递一个Time maxOutOfOrderness。不知道为什么要使用忽略 maxOutOfOrderness 的实现来覆盖其 getCurrentWatermark 方法。

public static void main(String[] args) throws Exception {

    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    DataStream<Event> events = ...

    events
        .assignTimestampsAndWatermarks(new TimestampsAndWatermarks(OUT_OF_ORDERNESS))
        .keyBy(e -> e.sessionId)
        .process(new RequestReponse())
        ...
}

public static class RequestReponse extends KeyedProcessFunction<KEY, Event, Long> {
    private ValueState<Long> requestTimeState;

    @Override
    public void open(Configuration config) {
        ValueStateDescriptor<Event> descriptor = new ValueStateDescriptor<>(
                "request time", Long.class);
        requestState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void processElement(Event event, Context context, Collector<Long> out) throws Exception {
        TimerService timerService = context.timerService();

        Long requestedAt = requestTimeState.value();
        if (requestedAt == null) {
            // haven't seen the request before; save its timestamp
            requestTimeState.update(event.timestamp);
            timerService.registerEventTimeTimer(event.timestamp + TIMEOUT);
        } else {
            // this event is the response
            // emit the time elapsed between request and response
            out.collect(event.timestamp - requestedAt);
        }
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext context, Collector<Long> out) throws Exception {
         //handle incomplete request/response events
    }
}

public static class TimestampsAndWatermarks extends BoundedOutOfOrdernessTimestampExtractor<Event> {
    public TimestampsAndWatermarks(Time t) {
        super(t);
    }

    @Override
    public long extractTimestamp(Event event) {
        return event.eventTime;
    }
}