Add a way to do intermediate aggregation

Adding basic aggregation support for keyed producer type source builders.

Main issues:
- is it much better to not involve the store?
- if using a store, can ScaldingStore be refactored to reduce the duplication in logic?
- if using a store, can the SourceBuilder changes be simplified?
- can we avoid specifying the reducer count here? pulling from options didn't seem to work
- can we avoid specifying the commutativity here? pulling from options didn't seem to work
- what's the best way to unit test this?
- need to get the storm passthrough