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

Betrieb: Verbindungsdatei, Summen mit Abgleich vor dem Schreiben, Laufhistorie

Browse Source
This commit is contained in:
wkh
2026-10-09 21:46:50 +02:00
parent fffb2e4961
commit baf09d3947
7 changed files with 390 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/datapipeline.rb
+141 -5
View File
@@ -7,6 +7,7 @@ require_relative 'datapipeline/validation'
require_relative 'datapipeline/layout'
require_relative 'datapipeline/watermark'
require_relative 'datapipeline/lock'
require_relative 'datapipeline/connections'
class DataPipeline
# plugins are loaded on first use, their gems too: pg (postgres), sqlite3 (sqlite), roo/caxlsx (excel),
@@ -30,21 +31,39 @@ class DataPipeline
# read: source rows, written: body rows, output: rows to the target incl. header/trailer,
# rejected: rows dropped by on_error :skip/:collect, errors: first 100 of them,
# warnings: rows that failed a validate step with warn: true (kept), error: exception of an aborted run
Stats = Struct.new(:version, :read, :written, :output, :rejected, :errors, :warnings, :started_at, :finished_at,
:dry_run, :error, keyword_init: true) do
# totals: { 'column' => { read: Rational, written: Rational } } of the columns named by totals
Stats = Struct.new(:name, :version, :read, :written, :output, :rejected, :errors, :warnings, :started_at,
:finished_at, :dry_run, :error, :totals, keyword_init: true) do
def duration = finished_at && started_at ? (finished_at - started_at).round(3) : nil
def ok? = error.nil? && rejected.zero?
# stats.total('dmbtr') -> sum after all steps, stats.total('dmbtr', :read) -> sum of the source rows
def total column, side = :written
totals.to_h.dig(column.to_s, side)
end
def to_s
"read=#{read} written=#{written} rejected=#{rejected}#{warnings.to_i.positive? ? " warnings=#{warnings}" : ''} " \
"read=#{read} written=#{written} rejected=#{rejected}#{warnings.to_i.positive? ? " warnings=#{warnings}" : ''}" \
"#{totals.to_h.map { |col, sum| " #{col}=#{DataPipeline.decimal(sum[:read])}/#{DataPipeline.decimal(sum[:written])}" }.join} " \
"#{duration}s#{dry_run ? ' (dry run)' : ''}" \
"#{error ? " ABORTED: #{error.class}: #{error.message}" : ''}"
end
# one line of the run history
def to_h
{ name:, version:, started_at: started_at&.iso8601(3), finished_at: finished_at&.iso8601(3), duration:,
read:, written:, output:, rejected:, warnings:, dry_run: dry_run ? true : false, ok: ok?,
error: error && "#{error.class}: #{error.message}",
totals: totals.to_h.transform_values { |sum| sum.transform_values { |v| DataPipeline.decimal(v) } } }
end
end
# stops reading at the limit, not an error
class LimitReached < StandardError; end
# 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
@@ -73,6 +92,8 @@ class DataPipeline
preview: 10,
on_error: :abort,
lock: nil,
name: nil,
history: nil,
log_level: Logger::WARN
}.freeze
@@ -157,6 +178,29 @@ class DataPipeline
def self.reset_deprecations! = SETTINGS_LOCK.synchronize { @deprecations_logged = [] }
# see DataPipeline::Connections
def self.connections_file = Connections.file
def self.connections_file=(path)
Connections.file = path
end
# entries of a run history file (see history), oldest first; name: only the runs of this pipeline
def self.history file, name: nil
return [] unless File.exist?(file)
entries = File.foreach(file).filter_map { |line| JSON.parse(line) unless line.strip.empty? }
name ? entries.select { |entry| entry['name'] == name.to_s } : entries
end
# exact decimal text of a sum: 1234.5 -> '1234.5', 1/3 -> '0.333333'
def self.decimal value
return value.to_s unless value.is_a?(Rational)
return value.to_i.to_s if value.denominator == 1
places = (1..10).find { |n| (value * 10**n).denominator == 1 } || 6
format("%.#{places}f", value)
end
# DataPipeline.new(verbose: true) { ... } overrides the defaults for this pipeline only
def initialize(**settings, &block)
@settings = DataPipeline.settings.merge(settings)
@@ -237,6 +281,43 @@ class DataPipeline
@settings[:lock] = file
end
# name of the pipeline in the stats and the run history; etl sets the script name as default
def name value
@settings[:name] = value
end
# history 'data/runs.ndjson' -> one json line per run (also aborted and dry runs), see DataPipeline.history
def history file
@settings[:history] = file
end
#
# totals 'dmbtr', 'menge' -> sums of these columns over the source rows and over the rows after all steps
# (stats.totals, stats.total('dmbtr')); empty and non numeric values are left out
#
def totals *columns
@totals = columns.flatten.map(&:to_s)
end
#
# reconcile { |stats| 'Summe weicht ab' if stats.total('dmbtr', :read) != stats.total('dmbtr') }
#
# runs after all steps, before writing (streaming: after the last row); a message (or array of messages)
# aborts the run with ReconcileError, nil means ok
#
def reconcile(&block)
raise ArgumentError, 'reconcile: block missing' unless block
(@reconcile ||= []) << block
end
# connection(:pds, query: '...') -> container of the type in the connections file, options of the call win
def connection name, **options, &block
settings = Connections.fetch(name)
type = settings.fetch(:type) { raise ArgumentError, "connection #{name.inspect}: type missing" }
container(type, **settings.except(:type), **options, &block)
end
# see DataPipeline::Watermark
def watermark name, column:, file:, initial: nil
@watermark = Watermark.new(name, column:, file:, initial:)
@@ -244,10 +325,21 @@ class DataPipeline
# only a missing plugin is reported as unsupported, errors while loading it
# (missing gem) or from the plugin itself (connection, options) are passed through
#
# container :postgres, connection: :pds, table: 'x' -> options from the connections file, the call wins
#
def container type, **options, &block
klass = DataPipeline.plugin(type)
raise "Unsupported storage type or class missing: #{type.to_s.capitalize}Storage" unless klass
if (name = options.delete(:connection))
settings = Connections.fetch(name)
if settings[:type] && settings[:type].to_s != type.to_s
raise ArgumentError, "connection #{name.inspect} is #{settings[:type]}, not #{type}"
end
options = settings.except(:type).merge(options)
end
klass.new(**options, &block)
end
@@ -371,8 +463,9 @@ class DataPipeline
end
def run
@stats = Stats.new(version: @settings[:version], read: 0, written: 0, output: 0, rejected: 0, errors: [], warnings: 0,
started_at: Time.now, dry_run: @settings[:dry_run])
@stats = Stats.new(name: @settings[:name], version: @settings[:version], read: 0, written: 0, output: 0, rejected: 0,
errors: [], warnings: 0, started_at: Time.now, dry_run: @settings[:dry_run],
totals: @totals&.to_h { |col| [col, { read: 0r, written: 0r }] })
@logger.info "start#{" #{@settings[:version]}" if @settings[:version]}#{@settings[:dry_run] ? ' (dry run)' : ''}"
@preview = []
raise ArgumentError, 'read storage missing' unless @source
@@ -444,6 +537,7 @@ class DataPipeline
self.class.runs << @stats
print_preview if @stats.dry_run && @stats.error.nil?
@logger.info "finish #{@stats}"
append_history
@on_finish.each do |hook|
hook.call(@stats)
rescue StandardError => e
@@ -451,6 +545,41 @@ class DataPipeline
end
end
# one json line per run, appended under an flock (parallel runs of other pipelines)
def append_history
return unless @settings[:history]
FileUtils.mkdir_p(File.dirname(@settings[:history]))
File.open(@settings[:history], 'a') do |file|
file.flock(File::LOCK_EX)
file.puts JSON.generate(@stats.to_h)
end
rescue StandardError => e
@logger.error "history: #{e.class}: #{e.message}"
end
def add_total side, row
@stats.totals&.each do |col, sum|
value = row.key?(col) ? row[col] : row[col.to_sym]
number = to_number(value)
sum[side] += number if number
end
end
def to_number value
case value
when Float then value.finite? ? value.to_s.to_r : nil
when Numeric then value.to_r
when String then value.strip.match?(Layout::NUMBER) ? value.strip.to_r : nil
end
end
# messages of the reconcile blocks abort the run
def check_reconcile
messages = Array(@reconcile).flat_map { |block| Array(block.call(@stats)).compact }
raise ReconcileError, messages.join('; ') if messages.any?
end
def source_rows
data = @source.read
data = data.first(@settings[:limit]) if @settings[:limit]
@@ -475,6 +604,7 @@ class DataPipeline
data = source_rows
@stats.read = data.length
data.each { |row| @watermark.track(row) } if @watermark
data.each { |row| add_total(:read, row) } if @totals
dump 'Source#data', data
# prepend keeps its old timing: called before the body
@@ -489,6 +619,9 @@ class DataPipeline
body.each { |step| rows.concat apply(step, data) }
end
dump 'Transformation#data', rows
@stats.written = rows.length
rows.each { |row| add_total(:written, row) } if @totals
check_reconcile
ctx = Context.new(read: data.length, written: rows.length)
before.select { |t| t[:kind] == :header }.each { |t| results[t] = edge(t, ctx, 0) }
@@ -525,8 +658,10 @@ class DataPipeline
read += 1
@watermark&.track(row)
add_total(:read, row) if @totals
push(row, body, counters, nil) do |out|
written += 1
add_total(:written, out) if @totals
emit.call out
end
end
@@ -535,6 +670,7 @@ class DataPipeline
end
@stats.read = read
@stats.written = written
check_reconcile
ctx = Context.new(read:, written:)
after.each { |t| emit.call edge(t, ctx, chained? ? written : read) }