Class: TWPipeline::Stages::Evidence

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

Constant Summary collapse

BATCH =

Returns:

  • (::Integer) 
20_000

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
28
29
30
31
32
33
34
35
36
 
# File 'lib/twpipeline/stages/evidence.rb', line 10

def call(input, output)
  blocks = 0
  kept = 0
  dropped = 0
  batch = []

  flush = lambda do
    Parallel.map(batch, workers: resources.cores) { |records| judge(records) }.each do |verdict|
      blocks += 1
      verdict.each do |record|
        record[:ok] == false ? dropped += 1 : kept += 1
        output.puts(Jsonl.dump(record)) unless drop? && record[:ok] == false
      end
    end

    batch = []
  end

  each_block(input) do |records|
    batch << records
    flush.call if batch.length >= BATCH
  end

  flush.call if batch.any?

  {blocks: blocks, kept: kept, dropped: dropped}
end