Piece Piece map, aggregate, group-by

group_by.rb

# Using group-by
# no extra step definition is required
PiecePipe::Pipeline.new.
  source([{region: region}]).
  step(FetchPowerPlantsByRegion).
  group_by(:region).
  to_enum

map_aggregate.rb

# each output would be a region => [plant_1_radiation, plant_2_radiation, ...]
# Using Map-Aggregate
PiecePipe::Pipeline.new.
  source([{region: region}]).
  step(FetchPowerPlantsByRegion).
  step(KeyByRegion).
  step(HashAggregator).
  to_enum 


# Expects inputs to be a Hash with :region, :date and :calories.
# Output will look like: { :some_region => original_item_including_region }
class KeyByRegion < PiecePipe::MapStep
  def map(inputs)
    key = inputs[:region]
    val = inputs
    emit key, val
  end
end