如果我理解正确，你可以这样做：

dataPayloadSource.keyby("room").process(new CountNumbers()).flatMap(new MapDataPayloadToCorrespondObject())addSink(...);

// ...
public class CountNumbers extends KeyedProcessFunction<..>{
    private MapState<Integer, Integer> numberCountState;

    public void open(Configuration config){
        // initialize state in here
    }

    public void processElement(DataPayload dp){
        // for each numbers in the dp.counts, get the state value with numberCountState.get(..)
        // check it returns null, if yes, map does not have that key, initialize with 1
        // if not null, then get the current value from the map, increment by 1
        // update the mapstate
    } 
}

// ...
public class MapDataPayloadToCorrespondObject extends RichFlatMapFunction<...>{
    public void flatMap(...){
        // convert DataPayload to OutputObject
    }
}