wkh/datapipeline
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

datapipeline 3.0.0: transform, Parallelmodus, chained und String-Bedingungen entfernt

Browse Source
This commit is contained in:
wkh
2026-10-09 22:04:12 +02:00
parent 0b72635e08
commit a8b8381293
13 changed files with 117 additions and 347 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/datapipeline.rb
+26 -114
View File
@@ -64,28 +64,15 @@ class DataPipeline
# raised when a reconcile check fails
class ReconcileError < StandardError; end
EDGE_BEFORE = %i[prepend header].freeze
EDGE_AFTER = %i[append trailer].freeze
CHAINING = %i[map filter validate].freeze
ON_ERROR = %i[abort skip collect].freeze
MAX_ERRORS = 100
# removed in 3.0, logged once per process as a warning
DEPRECATIONS = {
transform: 'transform is deprecated and will be removed in 3.0: use map/filter, ' \
'header/trailer instead of processing: :prepend/:append',
parallel: 'parallel mode (several transform steps without chained) is deprecated and will be removed in 3.0: ' \
'set chained or use map',
condition: 'condition: as string (eval) is deprecated and will be removed in 3.0: use a lambda ->(row) { ... }'
}.freeze
# value conversions for cast, empty values become nil
CASTS = Validation::TYPES.merge(string: ->(v) { v.to_s }).freeze
DEFAULT_SETTINGS = {
version: nil,
streamed: false,
chained: false,
verbose: false,
dry_run: false,
limit: nil,
@@ -100,7 +87,6 @@ class DataPipeline
SETTINGS_LOCK = Mutex.new
@defaults = DEFAULT_SETTINGS
@plugins = {}
@deprecations_logged = []
#
# defaults for every following DataPipeline.new, e.g. from bin/etl
@@ -166,18 +152,6 @@ class DataPipeline
@runs ||= []
end
# true the first time a deprecation is reported in this process
def self.deprecation_once key
SETTINGS_LOCK.synchronize do
next false if @deprecations_logged.include?(key)
@deprecations_logged << key
true
end
end
def self.reset_deprecations! = SETTINGS_LOCK.synchronize { @deprecations_logged = [] }
# see DataPipeline::Connections
def self.connections_file = Connections.file
def self.connections_file=(path)
@@ -204,7 +178,7 @@ class DataPipeline
# DataPipeline.new(verbose: true) { ... } overrides the defaults for this pipeline only
def initialize(**settings, &block)
@settings = DataPipeline.settings.merge(settings)
@transformations = []
@steps = []
@targets = []
@on_finish = []
@logger = Logger.new($stderr, level: @settings[:log_level], progname: 'etl')
@@ -232,11 +206,6 @@ class DataPipeline
@settings[:streamed] = flag
end
# each body step gets the output of the previous step instead of the source rows
def chained flag = true
@settings[:chained] = flag
end
# run everything, write nothing (target, rejects, watermark); prints the first rows
def dry_run flag = true
@settings[:dry_run] = flag
@@ -343,29 +312,19 @@ class DataPipeline
klass.new(**options, &block)
end
#
# transform(options) { |row, index| ... }
# processing: :prepend | :normal (default) | :append
# condition: ruby expression as string (row is visible) or lambda { |row| ... }
#
def transform(options = {}, &block)
kind = options.fetch(:processing, :normal)
kind = :normal unless %i[prepend append].include?(kind)
add_step kind, options, &block
end
# map { |row, ctx| ... } -> hash, array of hashes or nil; turns chaining on
# map { |row, ctx| ... } -> hash, array of hashes or nil
# condition: ->(row) { ... } on map and filter: the step only for matching rows, the others are dropped
def map(options = {}, &block)
add_step :map, options, &block
end
# filter { |row, ctx| ... } -> keeps the row when the block is truthy; turns chaining on
# filter { |row, ctx| ... } -> keeps the row when the block is truthy
def filter(options = {}, &block)
add_step :filter, options, &block
end
#
# short steps, each one a map step (turns chaining on)
# short steps, each one a map step
#
# rename 'kunnr' => 'kundennr' -> new key at the same position
# select %w[belnr betrag] -> only these keys in this order, missing ones nil
@@ -419,7 +378,7 @@ class DataPipeline
# validate layout -> fields of a DataPipeline::Layout
# validate { |row| 'soll ungleich haben' if ... } -> nil, message or array of messages
#
# a failing row raises ValidationError, handled by on_error; turns chaining on
# a failing row raises ValidationError, handled by on_error
# warn: true -> only a warning in the log and stats.warnings, the row is kept
# ignore: %w[A B] -> messages for these fields are left out (known gaps)
#
@@ -427,7 +386,7 @@ class DataPipeline
warn = options.delete(:warn) || false
ignore = Array(options.delete(:ignore)).map(&:to_s)
rules ||= options # validate 'konto' => { ... } arrives as keywords
@transformations << { kind: :validate, block:, options: { rules:, warn:, ignore: } }
@steps << { kind: :validate, block:, options: { rules:, warn:, ignore: } }
end
def read storage, options = {}
@@ -450,7 +409,7 @@ class DataPipeline
end
#
# loads a table once, for lookups in map/transform instead of one query per row
# loads a table once, for lookups in map instead of one query per row
#
# lookup pg, query: 'select ...', params: [...], key: 'rechnung_dbid' -> { key => [rows] }
# lookup pg, query: 'select ...', key: 'id', unique: true -> { key => row }
@@ -472,10 +431,9 @@ class DataPipeline
(@lock = Lock.new(@settings[:lock])).acquire if @settings[:lock]
raise ArgumentError, 'write storage missing' if @targets.empty?
before = @transformations.select { |t| EDGE_BEFORE.include?(t[:kind]) }
after = @transformations.select { |t| EDGE_AFTER.include?(t[:kind]) }
body = @transformations - before - after
log_deprecations body
before = @steps.select { |t| t[:kind] == :header }
after = @steps.select { |t| t[:kind] == :trailer }
body = @steps - before - after
@settings[:streamed] ? run_streamed(before, body, after) : run_batch(before, body, after)
@@ -493,8 +451,11 @@ class DataPipeline
def add_step kind, options, &block
raise ArgumentError, "#{kind}: block missing" unless block
if options.key?(:condition) && !options[:condition].respond_to?(:call)
raise ArgumentError, "#{kind}: condition must be a lambda ->(row) { ... }"
end
@transformations << { kind:, block:, options: }
@steps << { kind:, block:, options: }
end
def value_of row, key
@@ -511,21 +472,6 @@ class DataPipeline
raise ArgumentError, "#{key}: #{value.inspect} is not #{type}"
end
# each deprecated feature once per process, with the logger of the pipeline
def log_deprecations body
used = []
used << :transform if @transformations.any? { |t| %i[prepend normal append].include?(t[:kind]) }
used << :parallel if !chained? && body.size > 1
used << :condition if @transformations.any? { |t| t[:options].is_a?(Hash) && t[:options][:condition].is_a?(String) }
used.each { |key| @logger.warn "deprecated: #{DEPRECATIONS[key]}" if DataPipeline.deprecation_once(key) }
end
# without any transformation the source is copied unchanged
def chained?
@settings[:chained] || @transformations.empty? ||
@transformations.any? { |t| CHAINING.include?(t[:kind]) }
end
def write_row row
@targets.each { |t| t[:storage].write_row(row) if t[:only].nil? || t[:only].call(row) }
end
@@ -607,33 +553,21 @@ class DataPipeline
data.each { |row| add_total(:read, row) } if @totals
dump 'Source#data', data
# prepend keeps its old timing: called before the body
results = {}.compare_by_identity
before.select { |t| t[:kind] == :prepend }.each { |t| results[t] = edge(t, nil, 0) }
rows = []
if chained?
counters = Array.new(body.size, 0)
data.each { |row| push(row, body, counters, data.length) { |out| rows << out } }
else
body.each { |step| rows.concat apply(step, data) }
end
counters = Array.new(body.size, 0)
data.each { |row| push(row, body, counters, data.length) { |out| rows << out } }
dump 'Transformation#data', rows
@stats.written = rows.length
rows.each { |row| add_total(:written, row) } if @totals
check_reconcile
# header is computed after the body, so it knows the counters
ctx = Context.new(read: data.length, written: rows.length)
before.select { |t| t[:kind] == :header }.each { |t| results[t] = edge(t, ctx, 0) }
# parallel mode keeps the old count of source rows for append
count = chained? ? rows.length : data.length
new_data = []
before.each { |t| add_rows new_data, results[t] }
before.each { |t| add_rows new_data, t[:block].call(ctx) }
new_data.concat rows
after.each { |t| add_rows new_data, edge(t, ctx, count) }
after.each { |t| add_rows new_data, t[:block].call(ctx) }
@stats.written = rows.length
dump 'Target#data', new_data
output(new_data) do |all|
@targets.each { |t| t[:storage].write(t[:only] ? all.select(&t[:only]) : all) }
@@ -642,13 +576,9 @@ class DataPipeline
# row by row from source.read { } to target.write_row, header only knows nothing yet
def run_streamed before, body, after
if !chained? && body.size > 1
raise ArgumentError, 'streamed: several transformations need chained (or map/filter)'
end
emit = ->(result) { output(add_rows([], result)) { |rows| rows.each { |row| write_row(row) } } }
before.each { |t| emit.call edge(t, Context.new, 0) }
before.each { |t| emit.call t[:block].call(Context.new) }
read = written = 0
counters = Array.new(body.size, 0)
@@ -673,7 +603,7 @@ class DataPipeline
check_reconcile
ctx = Context.new(read:, written:)
after.each { |t| emit.call edge(t, ctx, chained? ? written : read) }
after.each { |t| emit.call t[:block].call(ctx) }
@targets.each { |t| t[:storage].finalize }
end
@@ -688,11 +618,6 @@ class DataPipeline
end
end
# parallel mode: every step reads its own copy of the source rows
def apply step, rows
rows.each_with_index.flat_map { |row, index| safe_step_rows(step, row.dup, index, rows.length) }
end
# errors of a step for one row go to on_error
def safe_step_rows step, row, index, read
step_rows(step, row, index, read)
@@ -715,7 +640,7 @@ class DataPipeline
def step_name step
name = step[:options].is_a?(Hash) && step[:options][:name] || step[:kind]
"#{name}##{@transformations.index(step) + 1}"
"#{name}##{@steps.index(step) + 1}"
end
# rows produced by one step for one input row
@@ -743,26 +668,13 @@ class DataPipeline
step[:block].call(row, Context.new(read:, index:)) ? [row] : []
when :map
add_rows [], step[:block].call(row, Context.new(read:, index:))
else
add_rows [], step[:block].call(row, index)
end
end
def edge step, ctx, count
case step[:kind]
when :prepend then step[:block].call(nil, 0)
when :append then step[:block].call(nil, count)
else step[:block].call(ctx)
end
end
# string conditions are evaluated with the local variable row
# condition: ->(row) { ... } on map and filter
def condition? step, row
condition = step[:options].fetch(:condition, true)
return true if condition == true
return condition.call(row) if condition.respond_to?(:call)
eval(condition)
condition = step[:options][:condition]
condition.nil? || condition.call(row)
end
# a hash is one row, an array are several rows, nil is no row
lib/datapipeline/version.rb
+1 -1
View File
@@ -1,5 +1,5 @@
class DataPipeline
VERSION = '2.1.0'.freeze
VERSION = '3.0.0'.freeze
# kept for scripts that print the version (rake info:info); the history is in CHANGELOG.md
class Version