Class: TWPipeline::Stages::Ingest

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

Constant Summary

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 



8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/twpipeline/stages/ingest.rb', line 8

def call(input, output)
  kept = 0
  residue = 0
  gates = Hash.new(0)
  sink = residue_path&.then { |path|
    path.dirname.mkpath
    path.open("w")
  }

  per_host = Hash.new(0)

  reader(input) do |io|
    source.each_record(io) do |record|
      gates[record[:gate]] += 1
      next if capped?(per_host, record)

      if record[:keep]
        kept += 1
        output.puts(Jsonl.dump(record.except(:keep)))
      elsif sink
        residue += 1
        sink.puts(Jsonl.dump(record.except(:keep)))
      end
    end
  end

  sink&.close
  {kept: kept, residue: residue, gates: gates}
end