Hi folks! We're a small startup just beginning to build our product on Flink. We're trying to implement a local aggregation followed by a global aggregation using the DataStream API. It didn't seem like there was an immediately visible way to do that, so what we're doing is using

DatastreamUtils.reinterpretAsKeyedStream()

and using the current task index (mapped in a rich function) as the key for the local aggregation. I'm not really sure that's the right thing to do, or what the behavior will end up looking like in case of restarts since the state for one partition may end up at some other task index later. What's the best way to do a local aggregation in Flink?