会话窗口闪烁

Question

有人能帮我理解一下在flink中的窗口（会话）是什么时候和如何发生的吗？或者样品是如何加工的？

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

如果我说的不对，请纠正。

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

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");
}

}

谢谢Vicky

Answer 1

从您的描述来看，我认为您希望编写一个自定义的ProcessFunction，它由session_id键控。您将有一个valueState，在其中存储请求事件的时间戳。当您获得相应的响应事件时，您会计算增量并发出该值（使用session_id)并清除出状态。

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