Add output batching support to Intermediate flatMap/source

[johnynek]

In storm, batching up a big block of items into a single tuple has been shown to improve performance by reducing contention on the queues.

The same trick could improve throughput between intermediate flatMaps by not flattening the results, but instead emitting lists.