PiecePipe Examples: Map, Aggregation, & Group-By

PiecePipe is a library that allows you to write code by focusing on the flow of data through a pipeline of steps without worrying about the iteration and “glue code.” It also makes testing much easier by removing the need to test the “glue code.”

Back in June, Dave wrote a blog post about the PiecePipe gem, but he left out the usage of MapStep and HashedAggregator.

Here is an example of using MapStep and HashedAggregator to do a simple group-by.

I’ve also added a GroupByStep to Piece Pipe that makes this example unnecessary.

# 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

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

If you are looking to sum values or use some sort of aggregate of the values, you should create a custom HashedAggregator step. Group by is a specialized case, but I found myself wanting it each time I used PiecePipe.