[streaming] GroupedBatchReduce modified to batch by key, updated WindowReduce + Batch/Window reduce refactor