Add windowFlatMapWithState function to DistributedUtil.

Reimplement pileup flatmap functions using windowFlatMapWithState, which drops a bunch of redundant code. Remove original window flatmap function.