在Flink ProcessWindowFunction中检查会话持续时间的全局状态

Question

我是Fink新手，希望计算EventFormDTO流的键控总会话持续时间：

    {"id":1,"projectId":1,"eventTypeId":1,"sessionId":"session_1","status":"IN_QUEUQ","comment":"","del":false,"millSecs":1}
{"id":1,"projectId":1,"eventTypeId":1,"sessionId":"session_2","status":"IN_QUEUQ","comment":"","del":false,"millSecs":5}
{"id":1,"projectId":1,"eventTypeId":1,"sessionId":"session_1","status":"ON_GOING","comment":"","del":false,"millSecs":10}
{"id":1,"projectId":1,"eventTypeId":1,"sessionId":"session_2","status":"ON_GOING","comment":"","del":false,"millSecs":18}

在队列中表示会话开始，而继续表示会话结束。预期输出应为事件到达时每个键控的总持续时间。因此，上述数据的样本输出为

ts        duration
timestamp_1  0
timestamp_2  4   // session_1: 5-1
timestamp_3  14  //(10-1)+ (10-5)  both session_1 and session_2 are active, and then session_1 end
timestamp_4  13  //18-5   the session_1 has already end.

在我的实现中，我使用了一个ProcessWindowFun和一个全局的MapState来跟踪

 public static void main(String[] args) {
    final DataStream<EventFormDTO> stream = ...;
    stream.keyBy(new KeySelector<EventFormDTO, String>() {
        @Override
        public String getKey(EventFormDTO eventFormDTO) throws Exception {
            return eventFormDTO.getProjectId()+"-"+eventFormDTO.getEventTypeId();
        }
    }).window(TumblingProcessingTimeWindows.of(Time.seconds(30))).process(new EventProcessWindowFun()).print();
    try {
        log.info("Start application.");
        env.execute();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

@Data
@AllArgsConstructor
@NoArgsConstructor
public class EventFormDTO implements Serializable {
    private static final long serialVersionUID = 5034868557373901846L;
    Long  id;
    Integer projectId;
    Integer eventTypeId;
    String sessionId;
    String status;
    String comment;
    Boolean del;
    Long millSecs;
}

@Slf4j
public class EventProcessWindowFun extends ProcessWindowFunction<EventFormDTO, Tuple2<Long, Long>, String, TimeWindow> {
    //session_id --> startTime.

    final static MapStateDescriptor<String, Long> descriptor =
            new MapStateDescriptor<>(
                    "record", // the state name
                    BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO
            );

    @Override
    public void process(String s, Context context, Iterable<EventFormDTO> iterable, Collector<Tuple2<Long, Long>> collector) throws Exception {
        MapState<String, Long > keyedMeasure = context.globalState().getMapState(descriptor);
        log.info("obtain reference to Map success.. ");
        for(EventFormDTO event: iterable){
            String sessionId = event.getSessionId();
            Long timeStamp = event.getMillSecs();
            String status = event.getStatus();
            Long duration = 0L;
            if(keyedMeasure.contains(sessionId)){
                duration += timeStamp - keyedMeasure.get(sessionId);
                keyedMeasure.remove(sessionId);
            }else{
                keyedMeasure.put(sessionId, timeStamp);
            }

            collector.collect(new Tuple2(context.window().getEnd(), duration));
        }
    }

}

然而，在调试过程中，我无法得到我想要的。

每次调用EventProcessWindowFun时，globalstatekeyedMeasure都是新的对象，并且没有在上一个窗口中计算的任何数据。

因此，我想问

1. 如何在ProcessWindowFunction中获取globalState
2. ProcessWindowFunction是否适用于我的案例？还有其他更好的解决方案吗？谢谢

Answer 1

keyedMeasure永远不会包含来自状态的键。我相信你是想写信的

if (keyedMeasure.contains(sessionId)) {
    duration += timeStamp - keyedMeasure.get(sessionId);
    keyedMeasure.remove(sessionId);
} else {
    keyedMeasure.put(sessionId, timeStamp)
}

如果将sessionId作为键的一部分，逻辑似乎会变得更简单。然后可以使用ValueState而不是MapState。

此外，您可能希望使用RichFlatMapFunction或KeyedProcessFunction而不是使用windows来完成此操作。通过使用windows，您可以在一些可以通过连续流处理更自然地完成的事情上施加一种小型批处理。