Class: TWPipeline::Stages::Dedup

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

Constant Summary collapse

MASK =

Returns:

  • (::Integer) 
(1 << 62) - 1

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 



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

def call(input, output)
  seen = Set.new
  hosts = Hash.new { |memo, key| memo[key] = Set.new }
  read = 0
  written = 0
  repeats = 0

  Jsonl.each(input) do |record|
    read += 1
    key = fingerprint(record.fetch(:text, ""))
    hosts[key] << record[:host] if record[:host] && hosts[key].size < ubiquity

    if seen.add?(key)
      written += 1
      output.puts(Jsonl.dump(record))
    else
      repeats += 1
      output.puts(Jsonl.dump(reject(record, [duplicate]))) unless drop?
    end
  end

  boilerplate = hosts.count { |_, group| group.size >= ubiquity }
  {read: read, written: written, duplicates: repeats, types: seen.size, boilerplate_types: boilerplate}
end