Class: TWPipeline::MapStage

Inherits:
Stage
  • Object
show all
Defined in:
lib/twpipeline/stage.rb,
sig/twpipeline.rbs

Constant Summary

Constants inherited from Stage

Stage::REGISTRY

Instance Attribute Summary

Attributes inherited from Stage

#options, #policy, #resources

Instance Method Summary collapse

Methods inherited from 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 



55
56
57
 
# File 'lib/twpipeline/stage.rb', line 55

def call(input, output)
  Parallel.pipe(input, output, workers: resources.cores) { |line| process(line) }
end

#process(line) ⇒ ::String?

Parameters:

  • line (::String)

Returns:

  • (::String, nil) 


59
60
61
62
63
64
65
66
67
 
# File 'lib/twpipeline/stage.rb', line 59

def process(line)
  stripped = line.strip
  return nil if stripped.empty?

  record = transform(Jsonl.parse(stripped))
  return nil if record.nil? || (drop? && record[:ok] == false)

  Jsonl.dump(record)
end

#transform(record) ⇒ record?

Parameters:

  • (record)

Returns:

  • (record, nil)

Raises:

  • (NotImplementedError) 


69
 
# File 'lib/twpipeline/stage.rb', line 69

def transform(record) = raise NotImplementedError