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 25554b6e7a
13 changed files with 117 additions and 347 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+8
View File
@@ -2,6 +2,14 @@
Framework `datapipeline` (Gem). Neueste Version oben.
## 3.0.0 – 2026-10-09
- **Breaking:** `transform` (auch `processing: :prepend`/`:append`), der Parallelmodus, `chained` und `condition:` als
String (`eval`) entfernt; alle Schritte laufen verkettet. Umstellung: `map`, `header`, `trailer`,
`condition: ->(row) { … }` (README 7.1)
- `condition:` muss ein Lambda sein, sonst `ArgumentError` beim Aufbau der Pipeline
- Hinweise `WARN deprecated` und `test/core/compat_test.rb` entfernt
## 2.1.0 – 2026-10-09
- Sperre gegen parallele Läufe: `lock 'datei'` in der Pipeline, `etl --lock=datei` für das ganze Skript; ein zweiter
Gemfile.lock
+1 -1
View File
@@ -1,7 +1,7 @@
PATH
remote: .
specs:
datapipeline (2.1.0)
datapipeline (3.0.0)
csv (~> 3.3)
logger (~> 1.6)
README.md
+43 -42
View File
@@ -1,6 +1,6 @@
# DataPipeline – ETL-Framework in Ruby
Stand: 2026-10-09 · Gem `datapipeline` 2.1.0 ([CHANGELOG.md](CHANGELOG.md)) · Ruby 3.4
Stand: 2026-10-09 · Gem `datapipeline` 3.0.0 ([CHANGELOG.md](CHANGELOG.md)) · Ruby 3.4
`DataPipeline` liest Daten aus einer Quelle, schickt jede Zeile durch eine Kette von Schritten (prüfen, filtern,
umbauen, anreichern, abgleichen) und schreibt das Ergebnis in ein oder mehrere Ziele. Eine Pipeline ist ein kurzes
@@ -15,7 +15,7 @@ Festbreiten-Dateien und REST-APIs.
4. [Plugins](#4-plugins) – mitgelieferte Quellen und Ziele, eigene Plugins, Abhängigkeiten
5. [Satzarten-Layouts](#5-satzarten-layouts) – Dateien mit Kopf-, Positions- und Schlusssätzen
6. [Betrieb](#6-betrieb) – Kommandozeile, Verbindungen, Sperre, Abgleich, Historie, Benachrichtigung, Fehlersuche
7. [Veraltetes und Umstieg](#7-veraltetes-und-umstieg) – von 2.0 auf 2.1, `transform` und Parallelmodus
7. [Umstieg](#7-umstieg) – von 2.x auf 3.0 (`transform` entfernt), von 2.0 auf 2.1
8. [Entwicklung](#8-entwicklung) – Aufbau, Tests, Gem bauen
9. [Ausblick](#9-ausblick)
@@ -783,7 +783,6 @@ Tippfehler fallen also sofort auf.
| --- | --- |
| `streamed(flag = true)` | Zeile für Zeile statt Batch (3.3) |
| `verbose(flag = true)` | Quelle, Ergebnis und Ziel im Batch als JSON auf `$stdout` |
| `chained(flag = true)` | nur für `transform` nötig (7.2) |
| `configure(opts)` | mehrere Einstellungen als Hash, z. B. `configure(streamed: true, version: '1.0')` |
| `run` | führt die Pipeline aus und liefert `DataPipeline::Stats` |
| `DataPipeline.new(**settings) { … }` | Einstellungen nur für diese Pipeline, z. B. `new(limit: 10)` |
@@ -918,8 +917,8 @@ Dry-Run wird die Fehlerdatei nicht geschrieben.
| `to_s` | `read=9 written=4 rejected=4 0.097s`; mit Warnungen `warnings=n`, mit `totals` `spalte=quelle/ergebnis` |
| `to_h` | alles als Hash, wie in der Laufhistorie |
Der Logger schreibt `start` und `finish` (Info), jede verworfene Zeile und Warnung (Warnung), veraltete Aufrufe
einmal je Prozess (Warnung, 7.2) und einen Abbruch (Fehler). `on_finish` läuft nach jedem Lauf, auch nach einem
Der Logger schreibt `start` und `finish` (Info), jede verworfene Zeile und Warnung (Warnung) und einen
Abbruch (Fehler). `on_finish` läuft nach jedem Lauf, auch nach einem
Abbruch; ein Fehler im Hook wird nur geloggt und ändert weder Ergebnis noch Exit-Code.
```ruby
@@ -1507,53 +1506,56 @@ gelesen 1250 · geschrieben 1243 · verworfen 7 · Warnungen 0
| `LayoutError: TEXT: "…" longer than 30` | Wert passt nicht in das Feld einer Festbreiten-Datei | vorher `validate LAYOUT` mit `on_error`, oder den Wert kürzen |
| `Encoding::UndefinedConversionError: U+20AC from UTF-8 to ISO-8859-1` | Zeichen gibt es im Zielzeichensatz nicht | `Windows-1252` statt ISO-8859-1, oder `replace: '?'` |
| `connection :x not in config/connections.yml (a, b)` | Name falsch oder Datei nicht gefunden | Namen prüfen; Pfad relativ zum Arbeitsverzeichnis, sonst `connections_file` setzen |
| `streamed: several transformations need chained (or map/filter)` | mehrere `transform` im Streaming | `map` verwenden (7.2) |
| `WARN deprecated: …` | veralteter Aufruf | umstellen, siehe 7.2 |
| `NoMethodError: undefined method 'transform'` (oder `chained`) | Skript für 2.x | auf `map`, `header`, `trailer` umstellen (7.1) |
| `map: condition must be a lambda ->(row) { ... }` | `condition:` als String aus 2.x | `condition: ->(row) { … }` (7.1) |
Mehr sehen: `--verbose` (Log-Level Debug), `verbose` in der Pipeline (Daten als JSON), `--dry-run --limit=20` für einen
schnellen Blick auf das Ergebnis.
---
## 7. Veraltetes und Umstieg
## 7. Umstieg
### 7.1 Von 2.0 auf 2.1
### 7.1 Von 2.x auf 3.0
2.1 ist abwärtskompatibel; bestehende Skripte laufen ohne Änderung. Anders als bisher:
3.0 entfernt das alte Ausführungsmodell: `transform`, den Parallelmodus, `chained` und `condition:` als String
(`eval`). Alle Schritte laufen jetzt verkettet, jeder bekommt die Ausgabe des vorigen. Alles andere ist unverändert;
ein Skript, das unter 2.1 ohne `WARN deprecated` lief, läuft auch unter 3.0.
| Thema | 2.0 | 2.1 |
| 2.x | 3.0 |
| --- | --- |
| `transform { \|row, index\| … }` | `map { \|row, ctx\| … }`, Index als `ctx.index` |
| `transform(processing: :prepend) { \|_, _\| kopf }` | `header { \|ctx\| kopf }` |
| `transform(processing: :append) { \|_, n\| schluss }` | `trailer { \|ctx\| schluss }`, Anzahl als `ctx.written` |
| `transform(condition: "row['a'] == 1") { … }` | `map(condition: ->(row) { row['a'] == 1 }) { … }` |
| `chained`, `configure(chained: true)` | weglassen |
Worauf beim Umstellen zu achten ist:
- **Parallelmodus:** Ohne `map`, `filter`, `validate` und `chained` bekam unter 2.x jeder `transform` eine eigene
Kopie der Quellzeilen, und die Ergebnisse wurden hintereinander angehängt. Zwei `transform` auf 2 Quellzeilen
ergaben 4 Zeilen, zwei `map` ergeben 2 Zeilen mit beiden Änderungen. Wer wirklich jede Zeile mehrfach ausgeben will,
gibt in einem `map` ein Array zurück: `map { |r| [mit_b(r), mit_c(r)] }`.
- **Kopfsatz:** `transform(processing: :prepend)` lief vor den Daten, `header` läuft im Batch erst danach (steht aber
vorne). Das zählt nur, wenn der Block etwas anderes tut, als eine Zeile zu liefern.
- **Schlusssatz:** `processing: :append` bekam im Parallelmodus die Anzahl der Quellzeilen; `ctx.written` ist die
Anzahl der Datenzeilen nach allen Schritten, `ctx.read` die der Quellzeilen.
- **Bedingung:** Ein String in `condition:` bricht schon beim Aufbau der Pipeline mit `ArgumentError` ab.
Prüfen lässt sich die Umstellung mit `etl run skript.rb --dry-run` vorher und nachher: Die Zusammenfassung und die
ersten Zeilen sollten gleich sein.
### 7.2 Von 2.0 auf 2.1
2.1 war abwärtskompatibel. Anders als in 2.0:
| Thema | 2.0 | ab 2.1 |
| --- | --- | --- |
| Postgres, `write_row` im Streaming | jede Zeile sofort, ohne Transaktion | je 1000 Zeilen in einer Transaktion; der Rest in `finalize`/`close`, auch nach einem Abbruch |
| CSV lesen mit UTF-8-BOM | BOM steht im ersten Spaltennamen | BOM wird übersprungen |
| `Stats#to_s` | – | mit `totals` zusätzlich `spalte=quelle/ergebnis` |
| `transform`, Parallelmodus, `condition:` als String | ohne Hinweis | `WARN deprecated: …` einmal je Prozess |
| `etl` ohne `name` | – | `stats.name` = Skriptname |
### 7.2 `transform` und Parallelmodus
`transform`, der Parallelmodus und `condition:` als String (`eval`) sind veraltet und entfallen in 3.0. Ab 3.0
laufen alle Schritte verkettet, wie heute schon, sobald `map`, `filter` oder `validate` vorkommt.
| Alt | Neu |
| --- | --- |
| `transform { \|row, index\| … }` | `map { \|row, ctx\| … }`, Index als `ctx.index` |
| `transform(processing: :prepend) { \|_, _\| kopf }` | `header { \|ctx\| kopf }` |
| `transform(processing: :append) { \|_, n\| schluss }` | `trailer { \|ctx\| schluss }` mit `ctx.written` |
| `transform(condition: "row['a'] == 1")` | `map(condition: ->(row) { row['a'] == 1 })` |
| mehrere `transform` ohne `chained` (Parallelmodus) | mehrere `map`; wer wirklich jede Zeile mehrfach ausgeben will, gibt in einem `map` ein Array zurück |
Bis dahin gilt für alte Skripte:
- **Parallelmodus** (Default ohne `map`/`filter`/`validate` und ohne `chained`): Jeder `transform` bekommt eine eigene
Kopie der Quellzeilen, die Ergebnisse aller Schritte werden hintereinander angehängt. Beispiel mit 2 Quellzeilen und
`transform { |r| r.merge('b' => 1) }` + `transform { |r| r.merge('c' => 2) }`: parallel 4 Zeilen (2× mit `b`, dann
2× mit `c`), verkettet 2 Zeilen mit `b` und `c`. Im Streaming sind mehrere `transform` nur verkettet erlaubt.
- **`transform(processing: :prepend)`** wird mit `(nil, 0)` vor den Daten aufgerufen, **`:append`** mit `(nil, n)`
nach den Daten (parallel n = Quellzeilen, verkettet n = Datenzeilen).
- **`condition:` als String** wird mit `eval` ausgewertet, die Variable `row` ist sichtbar.
`test/core/compat_test.rb` hält die Ausgabe des alten Modells fest, damit es sich bis 3.0 nicht ändert.
---
## 8. Entwicklung
@@ -1577,13 +1579,13 @@ Bis dahin gilt für alte Skripte:
### 8.2 Tests
`bundle exec rake test` startet die Minitest-Suite in [test/](test/), rund 205 Fälle. Einzelne Datei:
`bundle exec rake test` startet die Minitest-Suite in [test/](test/), rund 190 Fälle. Einzelne Datei:
`bundle exec rake test TEST=test/core/steps_test.rb`.
| Ordner | Inhalt |
| --- | --- |
| `test/plugins` | je Plugin das gemeinsame Testkit `test/support/storage_contract.rb` (Schreiben und Zurücklesen, zeilenweise, als Quelle und Ziel) plus eigene Fälle; REST gegen einen lokalen Testserver; Laden der Gems erst bei Nutzung |
| `test/core` | DSL, Schritte und Kurzschritte, Batch und Streaming, Fehlerbehandlung, Layouts, Sperre, Verbindungen, Summen, Historie, Wasserzeichen; `compat_test.rb` hält das alte `transform`-Modell fest |
| `test/core` | DSL, Schritte und Kurzschritte, Batch und Streaming, Fehlerbehandlung, Layouts, Sperre, Verbindungen, Summen, Historie, Wasserzeichen |
| `test/cli` | `bin/etl` als Prozess: Parameter, Dry-Run, Sperre, Historie, Exit-Codes |
| `test/examples` | die Beispiele aus Kapitel 2: Code in der Doku = Datei, Lauf und Ergebnis; REST gegen die Demo-API |
@@ -1604,7 +1606,7 @@ Ein eigenes Plugin lässt sich mit `MemoryStorage` aus `test/test_helper.rb` als
### 8.3 Gem und Versionen
- Version in [lib/datapipeline/version.rb](lib/datapipeline/version.rb), Änderungen in [CHANGELOG.md](CHANGELOG.md)
(neueste oben). Neue Funktionen erhöhen die mittlere Stelle, Breaking Changes die erste (3.0 entfernt `transform`).
(neueste oben). Neue Funktionen erhöhen die mittlere Stelle, Breaking Changes die erste (wie 3.0, 7.1).
- Das `Gemfile` bindet das Gem über `gemspec` ein und ergänzt die Gruppen `plugins` (`pg`, `sqlite3`, `roo`,
`caxlsx`, `http`) und `test`; `Gemfile.lock` ist versioniert. Ohne Plugin-Gems: `bundle config set without plugins`.
- Gem bauen: `gem build datapipeline.gemspec`.
@@ -1625,5 +1627,4 @@ Offene Punkte, Reihenfolge = empfohlene Umsetzung:
4. **Slack:** fertige Meldung aus den Stats (`notify slack, on: :problems`), Timeout und Wiederholung bei `429`,
Maskierung von `&`, `<`, `>`, Incoming Webhooks.
5. **Weitere Plugins:** SFTP-Ablage.
6. **3.0:** `transform`, Parallelmodus und `condition:` als String entfernen.
7. **Veröffentlichung:** eigenes Remote-Repository, Lizenz festlegen, Versionen taggen.
6. **Veröffentlichung:** eigenes Remote-Repository, Lizenz festlegen, Versionen taggen.
lib/datapipeline.rb
+24 -112
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
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
test/cli/etl_cli_test.rb
+4 -4
View File
@@ -14,8 +14,8 @@ class EtlCliTest < PipelineTest
"#{dir}/#{name}"
end
# like import.*.rb: own OptionParser for --from/--till, transform api
def test_legacy_script_gets_from_till_and_limit
# scripts with an own OptionParser for --from/--till
def test_script_with_own_option_parser_gets_from_till_and_limit
with_tmpdir do |dir|
path = script(dir, 'legacy.rb', <<~RUBY)
require 'optparse/date'
@@ -24,8 +24,8 @@ class EtlCliTest < PipelineTest
File.write('args.txt', "\#{opts[:from]} \#{opts[:till]}")
DataPipeline.new do
read container(:csv, file: 'in.csv')
transform(processing: :prepend) { |_, _| { 'h' => 'HEAD' } }
transform { |r, i| r.merge('i' => i) }
header { { 'h' => 'HEAD' } }
map { |r, ctx| r.merge('i' => ctx.index) }
write container(:csv, file: 'out.csv', csv: { headers: false, col_sep: '|' })
end.run
RUBY
test/core/chained_test.rb
+21 -15
View File
@@ -4,16 +4,22 @@ require_relative '../test_helper'
class ChainedTest < PipelineTest
ROWS = [{ 'a' => 1 }, { 'a' => 2 }, { 'a' => 3 }].freeze
def test_chained_transform_gets_previous_output
out = output(ROWS) { chained; transform { |r| r.merge('b' => 1) }; transform { |r| r.merge('c' => 2) } }
def test_each_step_gets_the_previous_output
out = output(ROWS) { map { |r| r.merge('b' => 1) }; map { |r| r.merge('c' => 2) } }
assert_equal ROWS.map { |r| r.merge('b' => 1, 'c' => 2) }, out
end
def test_one_to_many_then_index
out = output(ROWS) { chained; transform { |r| [r, r] }; transform { |r, i| r.merge('i' => i) } }
out = output(ROWS) { map { |r| [r, r] }; map { |r, ctx| r.merge('i' => ctx.index) } }
assert_equal (0...6).to_a, out.map { |r| r['i'] }
end
def test_nil_and_empty_array_drop_rows
assert_equal [{ 'a' => 1 }], output(ROWS) { map { |r| r['a'] == 1 ? r : nil } }
assert_equal ROWS, output(ROWS) { map { |r| [r, nil] } }
assert_equal [], output(ROWS) { map { [] } }
end
def test_map_and_filter
out = output(ROWS) do
map { |r| r.merge('d' => r['a'] * 2) }
@@ -23,12 +29,6 @@ class ChainedTest < PipelineTest
assert_equal [{ 'a' => 2, 'd' => 4, 'i' => 0 }, { 'a' => 3, 'd' => 6, 'i' => 1 }], out
end
def test_map_turns_chaining_on_for_transform
out = output(ROWS) { transform { |r| r.merge('b' => 1) }; map { |r| r.merge('c' => 1) } }
assert_equal 3, out.size
assert(out.all? { |r| r['b'] && r['c'] })
end
def test_header_and_trailer_with_context
out = output(ROWS) do
header { |c| { 'H' => c.read, 'W' => c.written } }
@@ -41,9 +41,14 @@ class ChainedTest < PipelineTest
assert_equal 6, out.size
end
def test_edge_order_follows_registration
out = output(ROWS) { header { { 'h' => 1 } }; transform(processing: :prepend) { { 'p' => 1 } }; map { |r| r } }
assert_equal [{ 'h' => 1 }, { 'p' => 1 }], out.first(2)
def test_several_headers_and_trailers_in_registration_order
out = output(ROWS) { header { { 'h' => 1 } }; trailer { { 't' => 1 } }; header { { 'h' => 2 } }; trailer { { 't' => 2 } } }
assert_equal [{ 'h' => 1 }, { 'h' => 2 }, *ROWS, { 't' => 1 }, { 't' => 2 }], out
end
def test_stable_order_of_many_steps
out = output(ROWS) { 20.times { |k| map { |r| r.merge("k#{k}" => k) } } }
assert_equal (0...20).to_a, out.first.values.drop(1)
end
def test_header_and_trailer_nil_and_arrays
@@ -51,12 +56,13 @@ class ChainedTest < PipelineTest
assert_equal ROWS + [{ 'x' => 1 }, { 'y' => 2 }], out
end
def test_condition_on_map
def test_condition_on_map_and_filter
assert_equal [{ 'a' => 1 }], output(ROWS) { map(condition: ->(r) { r['a'] == 1 }) { |r| r } }
assert_equal [{ 'a' => 3 }], output(ROWS) { filter(condition: ->(r) { r['a'] > 1 }) { |r| r['a'] == 3 } }
end
def test_chained_false
assert_equal 6, output(ROWS) { chained false; transform { |r| r }; transform { |r| r } }.size
def test_condition_must_be_a_lambda
assert_includes assert_raises(ArgumentError) { output(ROWS) { map(condition: "row['a'] == 1") { |r| r } } }.message, 'lambda'
end
def test_csv_end_to_end
test/core/compat_test.rb
-81
View File
@@ -1,81 +0,0 @@
require_relative '../test_helper'
#
# parallel mode (transform without chained/map/filter) must keep the output of v1.2.1,
# the version the CUDO importers were built on. expected values were recorded with v1.2.1.
#
class CompatTest < PipelineTest
ROWS = (1..6).map { |i| { 'id' => i, 'land' => i.even? ? 'DE' : 'AT' } }
def assert_output expected_json, &block
assert_equal expected_json, JSON.generate(output(ROWS, &block))
end
# pattern of import.customer.rb
def test_customer_pattern
assert_output '[{"SATYP":"FIRC","LFDNR":0},{"id":1,"land":"AT","lfdnr":"0000000001"},' \
'{"id":2,"land":"DE","lfdnr":"0000000002"},{"id":3,"land":"AT","lfdnr":"0000000003"},' \
'{"id":4,"land":"DE","lfdnr":"0000000004"},{"id":5,"land":"AT","lfdnr":"0000000005"},' \
'{"id":6,"land":"DE","lfdnr":"0000000006"},{"topic":"LARC","counter":"0000000006"}]' do
transform(processing: :prepend) { |_row, n| { 'SATYP': 'FIRC', 'LFDNR': n } }
transform(processing: :append) { |_row, n| { 'topic': 'LARC', 'counter': format('%010d', n) } }
transform { |row, _i| row.merge('lfdnr' => format('%010d', row['id'])) }
end
end
# pattern of import.posting.rb: closure counter, 1:n
def test_posting_pattern
expected = [{ SATYP: 'FIRC', LFDNR: 0 }]
lfdnr = 0
ROWS.each do |row|
expected << row.merge('lfdnr' => lfdnr += 1)
row['id'].times { |k| expected << { 'pos' => k, 'lfdnr' => lfdnr += 1 } }
end
expected << { SATYPE: 'LARC', SUMME: '0000000027' }
assert_output JSON.generate(expected) do
cnt = 1
transform(processing: :prepend) { |_row, _n| { 'SATYP': 'FIRC', 'LFDNR': 0 } }
transform do |row|
out = [row.merge('lfdnr' => cnt)]
cnt += 1
row['id'].times do |k|
out << { 'pos' => k, 'lfdnr' => cnt }
cnt += 1
end
out
end
transform(processing: :append) { |_row| { 'SATYPE': 'LARC', 'SUMME': format('%010d', cnt - 1) } }
end
end
def test_append_registered_before_prepend
assert_output '[{"h":0},{"id":1,"land":"AT"},{"id":2,"land":"DE"},{"id":3,"land":"AT"},' \
'{"id":4,"land":"DE"},{"id":5,"land":"AT"},{"id":6,"land":"DE"},{"t":6}]' do
transform(processing: :append) { |_, n| { 't' => n } }
transform { |r| r }
transform(processing: :prepend) { |_, n| { 'h' => n } }
end
end
def test_condition_string
assert_output '[{"id":2,"land":"DE","i":1},{"id":4,"land":"DE","i":3},{"id":6,"land":"DE","i":5}]' do
transform(condition: "row['land'] == 'DE'") { |r, i| r.merge('i' => i) }
end
end
def test_two_parallel_transforms
expected = ROWS.map { |r| r.merge('a' => 1) } + ROWS.map { |r| r.merge('b' => 1) }
assert_output JSON.generate(expected) do
transform { |r| r.merge('a' => 1) }
transform { |r| [r.merge('b' => 1), nil] }
end
end
def test_nil_and_empty_array
assert_output '[{"id":1,"land":"AT"}]' do
transform { |r| r['id'] > 3 ? nil : [] }
transform { |r| r['id'] == 1 ? r : nil }
end
end
end
test/core/operations_test.rb
+4 -4
View File
@@ -41,8 +41,8 @@ class OperationsTest < PipelineTest
assert_raises(RuntimeError) { run_pipeline(ROWS) { on_error :skip; header { raise 'h' }; map { |r| r } } }
end
def test_skip_in_parallel_mode
target, stats = run_pipeline(ROWS) { on_error :skip; transform { |r| r['id'] == '1' ? raise('x') : r } }
def test_skip_in_map
target, stats = run_pipeline(ROWS) { on_error :skip; map { |r| r['id'] == '1' ? raise('x') : r } }
assert_equal 4, target.output.size
assert_equal 1, stats.rejected
end
@@ -67,8 +67,8 @@ class OperationsTest < PipelineTest
assert_empty DataPipeline::Validation.errors({ 'a' => '' }, 'a' => { max: 1, type: :integer })
end
def test_validate_turns_chaining_on
target, = run_pipeline(ROWS) { on_error :skip; transform { |r| r.merge('x' => 1) }; validate RULES }
def test_validate_gets_the_output_of_the_previous_step
target, = run_pipeline(ROWS) { on_error :skip; map { |r| r.merge('x' => 1) }; validate RULES }
assert_equal 3, target.output.size
assert(target.output.all? { |r| r['x'] })
end
test/core/pipeline_test.rb
+3 -40
View File
@@ -1,17 +1,11 @@
require_relative '../test_helper'
# dsl, settings, plugin lookup, parallel mode
# dsl, settings, plugin lookup
class PipelineCoreTest < PipelineTest
ROWS = [{ 'a' => 1 }, { 'a' => 2 }, { 'a' => 3 }].freeze
def settings_of(pipeline) = pipeline.instance_variable_get(:@settings)
def test_nil_filters_rows
assert_equal [{ 'a' => 1 }], output(ROWS) { transform { |r| r['a'] == 1 ? r : nil } }
assert_equal ROWS, output(ROWS) { transform { |r| [r, nil] } }
assert_equal ROWS, output(ROWS) { transform(processing: :prepend) { nil }; transform { |r| r }; transform(processing: :append) { nil } }
end
def test_dsl_switches
assert_equal false, settings_of(DataPipeline.new { verbose false })[:verbose]
assert_equal true, settings_of(DataPipeline.new { verbose })[:verbose]
@@ -73,39 +67,8 @@ class PipelineCoreTest < PipelineTest
assert_equal Array.new(8) { |i| (i % 3) + 1 }, reads
end
def test_parallel_mode_is_side_by_side
out = output(ROWS) { transform { |r| r.merge('b' => 1) }; transform { |r| r.merge('c' => 2) } }
assert_equal 6, out.size
end
def test_parallel_mode_copies_rows_per_step
out = output(ROWS) { transform { |r| r['x'] = 1; r }; transform { |r| r.key?('x') ? nil : r } }
assert_equal 6, out.size
assert_equal [{ 'a' => 10 }, { 'a' => 20 }, { 'a' => 30 }], output(ROWS) { transform { |r| [r.transform_values! { |v| v * 10 }] } }
end
def test_stable_order_and_all_prepends_appends
out = output(ROWS) { 20.times { |k| transform { |r| r['a'] == 1 ? { 'k' => k } : nil } } }
assert_equal (0...20).to_a, out.map { |r| r['k'] }
out = output(ROWS) do
transform(processing: :prepend) { { 'h' => 1 } }
transform(processing: :prepend) { { 'h' => 2 } }
transform { |r| r }
transform(processing: :append) { { 't' => 1 } }
transform(processing: :append) { { 't' => 2 } }
end
assert_equal [{ 'h' => 1 }, { 'h' => 2 }, *ROWS, { 't' => 1 }, { 't' => 2 }], out
end
def test_append_count_parallel_vs_chained
assert_equal({ 'n' => 3 }, output(ROWS) { transform { |r| [r, r] }; transform(processing: :append) { |_, n| { 'n' => n } } }.last)
assert_equal({ 'n' => 6 }, output(ROWS) { chained; transform { |r| [r, r] }; transform(processing: :append) { |_, n| { 'n' => n } } }.last)
end
def test_conditions
assert_equal [{ 'a' => 1 }, { 'a' => 3 }], output(ROWS) { transform(condition: ->(r) { r['a'] != 2 }) { |r| r } }
assert_equal [{ 'a' => 2 }], output(ROWS) { transform(condition: "row['a'] == 2") { |r| r } }
def test_removed_dsl_of_version_2
%i[transform chained].each { |name| assert_raises(NoMethodError) { DataPipeline.new { send(name) { |r| r } } } }
end
def test_copy_without_steps
test/core/steps_test.rb
+1 -30
View File
@@ -1,6 +1,6 @@
require_relative '../test_helper'
# rename, select, cast, default and the deprecation warnings
# rename, select, cast, default
class StepsTest < PipelineTest
ROWS = [{ 'kunnr' => '1', 'betrag' => '12.50', 'datum' => '2026-10-09', 'land' => nil },
{ 'kunnr' => '2', 'betrag' => '', 'datum' => ' ', 'land' => 'DE' }].freeze
@@ -44,33 +44,4 @@ class StepsTest < PipelineTest
end
assert_equal [{ 'id' => '1', 'betrag' => 12.5 }, { 'id' => '2', 'betrag' => 0.0 }], rows
end
def deprecation_log(&block)
log = StringIO.new
DataPipeline.new do
logger Logger.new(log)
read MemoryStorage.new(rows: [{ 'a' => 1 }])
write MemoryStorage.new
instance_eval(&block)
end.run
log.string.lines.grep(/deprecated/)
end
def test_deprecations_once_per_process
lines = deprecation_log do
transform(condition: 'true') { |r| r }
transform { |r| r }
end
assert_equal 3, lines.size
assert lines.any? { |l| l.include?('transform is deprecated') }
assert lines.any? { |l| l.include?('parallel mode') }
assert lines.any? { |l| l.include?('condition: as string') }
assert_empty deprecation_log { transform { |r| r } }
end
def test_no_deprecation_for_map_filter_and_single_transform_in_chained_mode
assert_empty deprecation_log { map { |r| r }; filter(condition: ->(_r) { true }) { true } }
lines = deprecation_log { chained; transform { |r| r }; transform { |r| r } }
assert_equal 1, lines.size # only transform, no parallel mode
end
end
test/core/streaming_test.rb
+5 -14
View File
@@ -21,23 +21,14 @@ class StreamingTest < PipelineTest
assert_equal({ 'T' => 4, 'R' => 3 }, stream.last)
end
def test_stream_prepend_append
def test_stream_several_steps_one_to_many
out = output(ROWS, streamed: true) do
chained
transform(processing: :prepend) { |_, n| { 'p' => n } }
transform { |r| [r, r] }
transform(processing: :append) { |_, n| { 'n' => n } }
map { |r| [r, r] }
map { |r, ctx| r.merge('i' => ctx.index) }
trailer { |ctx| { 'n' => ctx.written } }
end
assert_equal({ 'p' => 0 }, out.first)
assert_equal({ 'n' => 6 }, out.last)
assert_equal 8, out.size
end
def test_stream_parallel_single_step_and_error_for_several
out = output(ROWS, streamed: true) { transform { |r| [r, r] }; transform(processing: :append) { |_, n| { 'n' => n } } }
assert_equal({ 'n' => 3 }, out.last)
error = assert_raises(ArgumentError) { output(ROWS, streamed: true) { transform { |r| r }; transform { |r| r } } }
assert_includes error.message, 'chained'
assert_equal (0...6).to_a, out.first(6).map { |r| r['i'] }
end
def test_stream_condition_and_copy
test/test_helper.rb
-1
View File
@@ -55,7 +55,6 @@ module PipelineHelper
DataPipeline.reset_settings!
DataPipeline.runs.clear
DataPipeline.params.clear
DataPipeline.reset_deprecations!
end
# runs a pipeline from rows into a MemoryStorage, returns [target, stats]