Flatten the stream and then do nested groupBy. This matches the mental model and is simple but
serializes the stream to a single thread which is bad for performance.

TODO: Add a version like this but with a ParallelObservable

@param stream
@return
540893
@SuppressWarnings("unused")
    private static Observable<GroupedObservable<TypeAndNameKey, Map<String, Object>>> aggregateUsingFlattenedGroupBy(Observable<GroupedObservable<InstanceKey, Map<String, Object>>> stream) {
        Observable<Map<String, Object>> allData = stream.flatMap(instanceStream -> {
            return instanceStream
                    .map((Map<String, Object> event) -> {
                        event.put("InstanceKey", instanceStream.getKey());
                        event.put("TypeAndName", TypeAndNameKey.from(String.valueOf(event.get("type")), String.valueOf(event.get("name"))));
                        return event;
                    })
                    .compose(is -> {
                        return tombstone(is, instanceStream.getKey());
                    });
        });

        Observable<GroupedObservable<TypeAndNameKey, Map<String, Object>>> byCommand = allData
                .groupBy((Map<String, Object> event) -> {
                    return (TypeAndNameKey) event.get("TypeAndName");
                });

        return byCommand
                .map(commandGroup -> {
                    Observable<Map<String, Object>> sumOfDeltasForAllInstancesForCommand = commandGroup
                            .groupBy((Map<String, Object> json) -> {
                                return json.get("InstanceKey");
                            }).flatMap(instanceGroup -> {
                                // calculate and output deltas for each instance stream per command
                                    return instanceGroup
                                            .takeUntil(d -> d.containsKey("tombstone"))
                                            .startWith(Collections.<String, Object> emptyMap())
                                            .map(data -> {
                                                if (data.containsKey("tombstone")) {
                                                    return Collections.<String, Object> emptyMap();
                                                } else {
                                                    return data;
                                                }
                                            })
                                            .buffer(2, 1)
                                            .filter(list -> list.size() == 2)
                                            .map(StreamAggregator::previousAndCurrentToDelta)
                                            .filter(data -> data != null && !data.isEmpty());
                                })
                            // we now have all instance deltas merged into a single stream per command
                            // and sum them into a single stream of aggregate values
                            .scan(new HashMap<String, Object>(), StreamAggregator::sumOfDelta)
                            .skip(1);

                    // we artificially wrap in a GroupedObservable that communicates the CommandKey this stream represents
                return GroupedObservable.from(commandGroup.getKey(), sumOfDeltasForAllInstancesForCommand);
            });
    }
1072521
public final static <T> CompletableFuture<List<T>> streamToCompletableFuture(final Stream<T> stream) {
        return CompletableFuture.completedFuture(stream.collect(Collectors.toList()));

    }
17038
public static final <A> Collection<A> toConcurrentLazyCollection(final Stream<A> stream) {
        return toConcurrentLazyCollection(stream.iterator());
    }
688492
public static <T> CloseableIterable<T> fromStream(Stream<T> stream) {
        return wrap(stream::iterator, stream);
    }
435846
public static <T, R> Map<R, T> mapBy(Stream<T> stream,
            Function<T, R> classifier) {
        return stream
                .collect(toMap(classifier, identity(), (o, o2) -> o2));
    }
1062275
public double sum(DoubleStream stream) {
        Callable<Double> callable = () -> stream.sum();
        return ThreadMethods.forkJoinExecution(callable, concurrencyConfiguration, stream.isParallel());
    }
1147871
@Override
    public StreamT<W,T> shuffle() {

        return (StreamT<W,T>) FoldableTransformerSeq.super.shuffle();
    }
1317616
public static <T, R> Map<R, T> merge(Stream<Map<R, T>> stream) {
        return stream.collect(HashMap::new, Map::putAll, Map::putAll);
    }
19967
public static <T> CloseableIterator<T> fromStream(Stream<T> stream) {
        return wrap(stream.iterator(), stream);
    }
937279
public <U> SimpleReactStream<U> from(final Stream<U> stream) {

        final Stream s = stream.map(it -> CompletableFuture.completedFuture(it));
        return construct(s);
    }
475967
public static <T> Stream<T> stream(final Iterable<T> source) {
        return new Stream<T>() {
            @Override
            public Iterator<T> iterator() {
                return source.iterator();
            }
        };
    }
854354
public static <T, K, V> Map<K, V> toMap( Stream<T> stream, Function<T, K> keySupplier, Function<T, V> valueSupplier ) {
        return stream.collect(Collectors.toMap(keySupplier, valueSupplier));
    }
1078062
public <K> Stream<T> distinctBy(final Function<? super T, K> keyMapper, final Predicate<? super Long> occurrencesFilter) {
        final Supplier<? extends Map<Keyed<K, T>, Long>> supplier = isParallel() ? Suppliers.<Keyed<K, T>, Long> ofConcurrentHashMap()
                : Suppliers.<Keyed<K, T>, Long> ofLinkedHashMap();

        return groupBy(Fn.<K, T> keyed(keyMapper), Collectors.counting(), supplier).filter(Fn.<Keyed<K, T>, Long> testByValue(occurrencesFilter))
                .map(Fn.<T, K, Long> kk());
    }
315711
@Override
	public SqlBatch paramStream(final Stream<?> stream) {
		if (this.stream == null) {
			this.stream = stream;
		} else {
			this.stream = Stream.concat(this.stream, stream);
		}
		return this;
	}
752731
public static final <K,V> Stream<Entry<K,V>> stream(K key, NavigableMap<K,V> map, boolean parallel)
    {
        return StreamSupport.stream(spliterator(key, map), parallel);
    }
612125
public <U> BaseSimpleReactStream<U> from(final DoubleStream stream) {

        return (BaseSimpleReactStream<U>) from(stream.boxed());

    }
403563
public List<Integer> get(String stream, List<Object> tuple, Collection<Tuple> anchors, Object rootId) {
        List<Integer> outTasks = new ArrayList<>();

        // get grouper, then get which task should tuple be sent to.
        Map<String, MkGrouper> componentCrouping = streamComponentGrouper.get(stream);
        if (componentCrouping == null) {
            // if the target component's parallelism is 0, don't need send to
            // them
            LOG.debug("Failed to get Grouper of " + stream + " when " + debugIdStr);
            return outTasks;
        }

        for (Entry<String, MkGrouper> ee : componentCrouping.entrySet()) {
            String targetComponent = ee.getKey();
            MkGrouper g = ee.getValue();

            if (GrouperType.direct.equals(g.gettype())) {
                throw new IllegalArgumentException("Cannot do regular emit to direct stream");
            }

            outTasks.addAll(g.grouper(tuple));
        }

        if (isDebug(anchors, rootId)) {
            LOG.info(debugIdStr + stream + " to " + outTasks + ":" + tuple.toString());
        }
        int num_out_tasks = outTasks.size();
        taskStats.send_tuple(stream, num_out_tasks);

        return outTasks;
    }