Class: TWPipeline::Stages::Count

Inherits:
TWPipeline::Stage show all
Defined in:
lib/twpipeline/stages/count.rb,
sig/twpipeline.rbs

Constant Summary collapse

DEFAULT_ORDERS =

Returns:

  • (::Array[::Integer]) 
[1, 2, 3, 4].freeze

Constants inherited from TWPipeline::Stage

TWPipeline::Stage::REGISTRY

Instance Attribute Summary

Attributes inherited from TWPipeline::Stage

#options, #policy, #resources

Instance Method Summary collapse

Methods inherited from TWPipeline::Stage

#drop?, #flag, #initialize, lookup, order, ordered, register, #reject, slug

Constructor Details

This class inherits a constructor from TWPipeline::Stage

Instance Method Details

#call(input, output) ⇒ Object 



10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
 
# File 'lib/twpipeline/stages/count.rb', line 10

def call(input, output)
  sinks = {}
  pipelines = orders.to_h { |order| [order, sink(order, sinks)] }
  records = 0

  Jsonl.each(input) do |record|
    records += 1
    units = units_of(record)
    part = part_of(record)
    orders.each do |order|
      emit(units, order) { |gram| pipelines.fetch(order).call(part ? "#{gram}\t#{part}" : gram) }
    end
  end

  results = close(sinks)
  output.puts(Jsonl.dump({records: records, orders: results}))
  {records: records, orders: results}
end