A better approach is to use DataStream API.

Use a FlatMap function to extract rows from list and then do the aggregation. Or use a MapFunction to caculate to only aggregate on a List rows.

These operations are not stateful so you are not using flink states. You also process each row individually so you do not need to set window at all.