From 0b72635e08c77d5ecdf81102739834d8bca7511a Mon Sep 17 00:00:00 2001 From: Wild Karl-Heinz Date: Fri, 9 Oct 2026 21:55:16 +0200 Subject: [PATCH] =?UTF-8?q?README=20neu=20gegliedert,=20Beispiel=20f=C3=BC?= =?UTF-8?q?r=20einen=20produktiven=20Lauf,=20Fehlersuche=20und=20Umstiegsh?= =?UTF-8?q?inweise?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 3 + README.md | 1491 ++++++++++++++++++---------- examples/buchungen_export.rb | 59 ++ examples/config/connections.yml | 8 + examples/data/buchungen.csv | 8 + test/examples/doc_examples_test.rb | 37 +- 6 files changed, 1070 insertions(+), 536 deletions(-) create mode 100644 examples/buchungen_export.rb create mode 100644 examples/config/connections.yml create mode 100644 examples/data/buchungen.csv diff --git a/CHANGELOG.md b/CHANGELOG.md index 8a1f8fb..ab29c46 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,9 @@ Framework `datapipeline` (Gem). Neueste Version oben. vor dem Schreiben, eine Meldung bricht mit `DataPipeline::ReconcileError` ab - Laufhistorie: `history 'datei'` bzw. `etl --history=datei`, eine JSON-Zeile je Lauf; `DataPipeline.history` liest sie; `name` des Laufs (über `etl` der Skriptname), `Stats#to_h` +- README neu gegliedert (Überblick, Beispiele, Pipelines schreiben, Plugins, Layouts, Betrieb, Umstieg, + Entwicklung) mit Inhaltsverzeichnis, Fehlersuche und Umstiegshinweisen; neues Beispiel + `examples/buchungen_export.rb` für einen produktiven Lauf ## 2.0.0 – 2026-10-08 diff --git a/README.md b/README.md index 2de293d..903fd79 100644 --- a/README.md +++ b/README.md @@ -1,39 +1,61 @@ -# DataPipeline – generisches ETL-Framework in Ruby +# DataPipeline – ETL-Framework in Ruby Stand: 2026-10-09 · Gem `datapipeline` 2.1.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 +Ruby-Skript; Quellen und Ziele sind austauschbare Plugins für PostgreSQL, SQLite, CSV, JSON/NDJSON, Excel, +Festbreiten-Dateien und REST-APIs. + +**Inhalt** + +1. [Überblick](#1-überblick) – was das Gem kann, Schnellstart, Grundbegriffe +2. [Beispiele](#2-beispiele) – sieben lauffähige Pipelines, vom Import bis zum produktiven Export +3. [Pipelines schreiben](#3-pipelines-schreiben) – DSL, Schritte, Prüfen, Fehler, Ergebnis +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 +8. [Entwicklung](#8-entwicklung) – Aufbau, Tests, Gem bauen +9. [Ausblick](#9-ausblick) + --- -## 1. Beschreibung +## 1. Überblick -`DataPipeline` ist ein schlankes ETL-Framework (Extract – Transform – Load) mit einer Ruby-DSL. Eine Pipeline -liest Daten aus einer Quelle, schickt jede Zeile durch eine Kette von Schritten (prüfen, filtern, umbauen, -anreichern) und schreibt das Ergebnis in ein oder mehrere Ziele. Quellen und Ziele sind austauschbare -**Storage-Plugins**: PostgreSQL, SQLite, CSV, JSON/NDJSON, Excel und REST-APIs. +### 1.1 Was das Gem kann -Die Pipeline beschreibt *was* passiert, die Plugins kapseln *wie* gelesen und geschrieben wird. Für den Betrieb -gibt es Fehlerdateien, Prüfregeln, Dry-Run, inkrementelle Läufe, Streaming für grosse Datenmengen und eine -Kommandozeile mit Exit-Codes. +| Bereich | Möglichkeiten | +| --- | --- | +| Lesen und Schreiben | PostgreSQL, SQLite, CSV, JSON/NDJSON, Excel, Festbreiten-Dateien, REST-APIs; eigene Plugins mit wenigen Methoden; mehrere Ziele je Pipeline, auch mit Auswahl je Ziel | +| Umbauen | `map` (1:1, 1:n, 1:0), `filter`, Kurzschritte `rename`, `select`, `cast`, `default`; Nachschlagetabellen mit `lookup`; Kopf- und Schlusssätze mit Zählern | +| Prüfen | Regeln je Spalte (Pflicht, Länge, Format, Werteliste, Typ), eigene Regeln als Block, Satzarten-Layouts; als Fehler oder nur als Warnung | +| Fehler | Abbruch, Zeile überspringen oder mit Grund in eine Fehlerdatei schreiben | +| Datenmengen | Batch im Speicher oder Streaming Zeile für Zeile mit konstantem Speicherbedarf; Postgres-Cursor und Schreiben in Paketen | +| Betrieb | Kommandozeile `etl` mit Exit-Codes, Dry-Run, Limit, inkrementelle Läufe (Wasserzeichen), Sperre gegen parallele Läufe, Upsert für wiederholbare Läufe, Summenabgleich vor dem Schreiben, Laufhistorie, Verbindungsdatei, Slack | +| Dateiformate | Zeichensatz (z. B. Windows-1252) für CSV und Festbreite, BOM für CSV; Satzarten-Dateien schreiben, prüfen und lesen | **Anwendungen:** Das Projekt `etl` nutzt das Gem für die PDS → SAP-Schnittstelle (CUDO) und für das Anlegen von -Rechnungen über die pds API; dort liegen auch die zugehörige Doku und Tests. +Rechnungen über die pds API; dort liegen auch deren Doku und Tests. -### 1.1 Schnellstart +### 1.2 Schnellstart + +Im Repository: ```sh mise install # Ruby 3.4.9 aus mise.toml bundle install -bundle exec rake test # Tests, lokales Postgres optional -bin/etl run examples/kunden_import.rb # Beispiel unten, Ergebnis in tmp/ +bundle exec rake test # Tests; Postgres-Tests nur mit lokalem Server (8.2) +bin/etl run examples/kunden_import.rb # Beispiel 2.1, Ergebnis in tmp/ bin/etl run examples/kunden_import.rb --dry-run # alles prüfen, nichts schreiben ``` -Das Framework ist das Gem `datapipeline`. In einer anderen Anwendung wird es so eingebunden: +In einer anderen Anwendung: ```ruby # Gemfile gem 'datapipeline', path: '../datapipeline' # oder git: '…' -gem 'pg' # nur die Gems der benötigten Plugins, siehe 2.6 +gem 'pg' # nur die Gems der benötigten Plugins (4.10) ``` ```sh @@ -41,7 +63,7 @@ bundle install bundle exec etl pipeline.rb --dry-run # das Kommando etl kommt mit dem Gem ``` -Eine Pipeline ist ein Ruby-Skript. Das kleinste sinnvolle Beispiel kopiert eine CSV-Datei nach JSON: +Die kleinste sinnvolle Pipeline kopiert eine CSV-Datei nach JSON: ```ruby require 'datapipeline' @@ -52,16 +74,38 @@ DataPipeline.new do end.run ``` -Zwischen `read` und `write` stehen die Schritte; sie laufen in der Reihenfolge, in der sie im Skript stehen: +### 1.3 Grundbegriffe -| Schritt | Wirkung | +| Begriff | Bedeutung | | --- | --- | -| `validate regeln` | prüft jede Zeile; Fehler gehen an `on_error` | -| `filter { \|row\| … }` | Zeile bleibt, wenn der Block wahr ist | -| `map { \|row, ctx\| … }` | gibt eine Zeile, mehrere (Array) oder keine (`nil`) zurück | -| `header { \|ctx\| … }`, `trailer { \|ctx\| … }` | Kopf- und Schlusssatz mit den Zählern `ctx.read`, `ctx.written` | +| Pipeline | `DataPipeline.new do … end`: eine Quelle, Schritte, ein oder mehrere Ziele; `run` führt sie aus | +| Zeile | ein Hash mit String-Schlüsseln, z. B. `{ 'kundennr' => '100001', 'name' => 'Müller GmbH' }` | +| Plugin, Storage | liest oder schreibt Zeilen (`container(:csv, …)`); dieselbe Klasse dient als Quelle und als Ziel | +| Schritt | `validate`, `filter`, `map` und die Kurzschritte; jede Zeile läuft der Reihe nach durch alle Schritte | +| Kopf-, Schlusssatz | `header`, `trailer`: Zeilen vor bzw. nach den Daten, z. B. mit der Anzahl Datensätze | +| Batch, Streaming | Batch lädt die Quelle ganz in den Speicher, Streaming (`streamed`) verarbeitet Zeile für Zeile | +| Lauf, Stats | ein Aufruf von `run`; das Ergebnis `DataPipeline::Stats` enthält Zähler, Fehler, Summen und Laufzeit | +| Satzart, Layout | Dateien mit verschiedenen Zeilenarten (Kopf, Position, Ende) und fester Feldliste je Art (Kapitel 5) | +| Wasserzeichen | der höchste verarbeitete Wert einer Spalte, damit der nächste Lauf nur Neues liest (6.5) | -### 1.2 Ausführliches Beispiel: Kundenimport +--- + +## 2. Beispiele + +Alle Beispiele liegen in [examples/](examples/), lesen die Daten in `examples/data/` und schreiben nach `tmp/` +(`--out=verzeichnis` ändert das). `test/examples/doc_examples_test.rb` führt jedes aus, prüft das Ergebnis und dass +der Code hier genau der Datei entspricht. + +| Beispiel | Zeigt | +| --- | --- | +| [2.1 Kundenimport](#21-kundenimport) | Prüfen, Nachschlagen, Fehlerdatei, zwei Ziele | +| [2.2 Datei mit Satzarten](#22-datei-mit-satzarten) | 1:n-Umbau, Kopf- und Schlusssatz, Layout | +| [2.3 Inkrementell und gestreamt](#23-inkrementell-und-gestreamt) | Wasserzeichen, Streaming | +| [2.4 Eigenes Plugin](#24-eigenes-plugin) | eine Quelle selbst schreiben | +| [2.5 REST-APIs](#25-rest-apis) | Paging, Wiederholung, Pakete, eine URL je Datensatz | +| [2.6 Produktiver Lauf](#26-produktiver-lauf) | Verbindungsdatei, Sperre, Kurzschritte, Summenabgleich, Upsert, Zeichensatz, Historie, Slack | + +### 2.1 Kundenimport Eine Kundenliste (`examples/data/kunden.csv`, 9 Zeilen) soll geprüft, mit Ländernamen angereichert und in eine SQLite-Datenbank geschrieben werden; EU-Kunden kommen zusätzlich in eine Excel-Liste. Fehlerhafte Zeilen sollen den @@ -135,20 +179,18 @@ DataPipeline.new do end.run ``` -Was die einzelnen Teile tun: - | Zeile | Erklärung | | --- | --- | | `DataPipeline.params.fetch(:out, 'tmp')` | Parameter von der Kommandozeile: `--out=verzeichnis` | | `lookup …, key: 'code', unique: true` | lädt die Länder einmal als Hash nach Ländercode; kein Zugriff je Zeile | | `on_error :collect, to: …` | jede Zeile, die in einem Schritt scheitert, kommt mit `_step` und `_error` in die Fehlerdatei | -| `validate 'kundennr' => { … }` | Regeln je Spalte: `required`, `format`, `max`, `min`, `in`, `type` (`:integer`, `:decimal`, `:date`; siehe 2.7) | +| `validate 'kundennr' => { … }` | Regeln je Spalte (3.5) | | `validate { \|row\| … }` | eigene Regel: ein Text (oder ein Array) ist ein Fehler, `nil` heisst in Ordnung | | `filter` | inaktive Kunden fallen weg, ohne Fehler | | `map` | baut die Zielzeile; `laender.fetch` kann nicht scheitern, weil `validate` das Land schon geprüft hat | | `write …` zweimal | jedes Ziel bekommt die Zeilen; `only:` wählt die Zeilen für die Excel-Liste | | `create: true, truncate: true` | SQLite legt die Tabelle beim ersten Schreiben an und leert sie; im Dry-Run passiert nichts | -| `on_finish` | läuft nach jedem Lauf, auch nach einem Abbruch; `stats` siehe 2.7 | +| `on_finish` | läuft nach jedem Lauf, auch nach einem Abbruch; `stats` siehe 3.7 | Ergebnis von `bin/etl run examples/kunden_import.rb`: @@ -172,15 +214,12 @@ Der Exit-Code ist `2`, weil Zeilen verworfen wurden; ein Abbruch ergäbe `1`, ei `--dry-run` läuft alles gleich, es entstehen aber weder Datenbank noch Excel- oder Fehlerdatei; stattdessen stehen die ersten Zeilen auf dem Bildschirm. -### 1.3 Weitere Beispiele +### 2.2 Datei mit Satzarten -Alle Beispiele liegen in [examples/](examples/), arbeiten mit den Daten in `examples/data/` und schreiben nach `tmp/`. -`test/examples/doc_examples_test.rb` führt sie aus und prüft, dass der Code hier mit den Dateien übereinstimmt. - -**Datei mit Satzarten** – Rechnungen aus verschachteltem JSON werden zu einem Kopfsatz mit Positionssätzen; Datei- -und Schlusssatz zählen die Datensätze. Das Layout legt Reihenfolge, Typ und Länge der Felder je Satzart fest; die -Datentypen `AN`, `N`, `DAT`, `TIM` und ihre Prüfung stehen in 2.9. Ohne `format: true` formatiert das Layout nicht, -deshalb bringt das `map` hier Datum und Beträge selbst ins Zielformat (`delete('-')`, `format('%.2f', …)`). +Rechnungen aus verschachteltem JSON werden zu einem Kopfsatz mit Positionssätzen; Datei- und Schlusssatz zählen die +Datensätze. Das Layout legt Reihenfolge, Typ und Länge der Felder je Satzart fest (Kapitel 5). Ohne `format: true` +formatiert das Layout nicht, deshalb bringt das `map` hier Datum und Beträge selbst ins Zielformat (`delete('-')`, +`format('%.2f', …)`). ```ruby @@ -248,8 +287,10 @@ POS|000005|R-1002|001|Jahreslizenz Software mit Wartung und Support|1200.00 ENDE|000005 ``` -**Inkrementell und gestreamt** – nur Bewegungen, die seit dem letzten erfolgreichen Lauf dazugekommen sind; jede -Zeile wird sofort geschrieben, der Speicherbedarf bleibt auch bei Millionen Zeilen klein. +### 2.3 Inkrementell und gestreamt + +Nur Bewegungen, die seit dem letzten erfolgreichen Lauf dazugekommen sind; jede Zeile wird sofort geschrieben, der +Speicherbedarf bleibt auch bei Millionen Zeilen klein. ```ruby @@ -288,8 +329,10 @@ end.run Der erste Lauf schreibt 4 Bewegungen und merkt sich `4`; der zweite schreibt nichts (`neu: 0`) und legt auch keine Datei an. Bricht ein Lauf ab, bleibt der alte Stand stehen. -**Eigenes Plugin** – eine Quelle, die alle CSV-Dateien eines Verzeichnisses liest. Ein Plugin braucht nur die -Methoden, die es anbietet (hier `read`); die Schnittstelle steht in 2.4. +### 2.4 Eigenes Plugin + +Eine Quelle, die alle CSV-Dateien eines Verzeichnisses liest. Ein Plugin braucht nur die Methoden, die es anbietet +(hier `read`); die Schnittstelle steht in 4.1. ```ruby @@ -343,9 +386,7 @@ zuerich,A-100,7 zuerich,C-300,1 ``` -Weitere Ausschnitte: REST-API nach Excel in 2.5 (REST ausführlich in 1.4), Benachrichtigung über Slack in 2.7. - -### 1.4 Beispiele mit REST-APIs +### 2.5 REST-APIs Die drei Beispiele laufen gegen eine kleine Demo-API ([examples/demo_api.rb](examples/demo_api.rb)), die ohne Internet und ohne weitere Gems auskommt. Sie verlangt ein Token, liefert Seiten und Folge-Links, nimmt Daten per POST und PUT an @@ -358,7 +399,7 @@ bin/etl run examples/rest_schreiben.rb bin/etl run examples/rest_zu_rest.rb ``` -Gegen eine echte API ändern sich nur `--url=…` und das Token (`DEMO_API_TOKEN`, in der Praxis aus der `.env`). +Gegen eine echte API ändern sich nur `--url=…` und das Token (`DEMO_API_TOKEN`, in der Praxis aus der Umgebung). | Endpunkt der Demo-API | Antwort | | --- | --- | @@ -544,257 +585,543 @@ archiviert: 5, Stand jetzt: 5 # erster Lauf: PUT /api/archiv/1 … /5 archiviert: 0, Stand jetzt: 5 # zweiter Lauf: keine Anfrage ``` +### 2.6 Produktiver Lauf + +Ein Export, wie er nachts per Cron läuft. Buchungszeilen mit Vorzeichen (`betrag`, negativ = Haben) werden geprüft, +in Soll/Haben-Kennzeichen und Betrag umgebaut und vor dem Schreiben abgeglichen: Die Summe darf sich durch die +Schritte nicht ändern, und jede Buchung muss ausgeglichen sein (Saldo 0). Ziel sind eine Buchhaltungs-Datenbank aus +der Verbindungsdatei und eine CSV-Datei im Zeichensatz des Zielsystems. + +```yaml +# Verbindungen der Beispiele. Die Datei wird mit ERB gelesen: Zugangsdaten und Pfade kommen aus der Umgebung, +# bei Postgres z. B. password: <%= ENV['PDS_PASSWORD'] %> +buchhaltung: + type: sqlite + database: <%= ENV.fetch('BUCHHALTUNG_DB', 'tmp/buchhaltung.sqlite3') %> + table: buchungen + create: true + upsert: [belnr, buzei] +``` + + +```ruby +# +# Produktiver Lauf: Buchungen prüfen, Summe und Saldo abgleichen, wiederholbar in die Buchhaltungs-Datenbank +# und als CSV in Windows-1252 für das Zielsystem schreiben; mit Sperre, Laufhistorie und Slack bei Problemen +# +# bin/etl run examples/buchungen_export.rb [--out=tmp] [--dry-run] +# +require 'fileutils' +require 'datapipeline' + +DATA = File.expand_path('data', __dir__) +out = DataPipeline.params.fetch(:out, 'tmp') +FileUtils.mkdir_p(out) + +# Zugangsdaten stehen nicht im Skript, sondern in der Verbindungsdatei (Default: config/connections.yml) +DataPipeline.connections_file = File.expand_path('config/connections.yml', __dir__) + +# Slack nur, wenn ein Token gesetzt ist; so fällt ein fehlender Token schon vor dem Lauf auf +slack = DataPipeline::Slack.new(channel: '#etl') if ENV['SLACK_TOKEN'] + +DataPipeline.new do + name 'buchungen_export' + lock "#{out}/.buchungen.lock" # läuft schon ein Export, bricht dieser ab (Exit-Code 3) + history "#{out}/runs.ndjson" # eine JSON-Zeile je Lauf, auch nach Abbruch + + read container(:csv, file: "#{DATA}/buchungen.csv", csv: { col_sep: ';' }) + + # 1. prüfen: ohne on_error bricht ein Fehler den Lauf ab, es entsteht keine halbe Buchung + validate 'belnr' => { required: true, format: /\A\d{10}\z/ }, + 'konto' => { required: true, format: /\A\d+\z/ }, + 'betrag' => { required: true, type: :decimal } + + # 2. umbauen mit Kurzschritten und einem map + default 'text' => 'ohne Text' + rename 'text' => 'buchungstext' + cast 'buzei' => :integer, 'datum' => :date, 'betrag' => :decimal + map do |row| + row.merge('shkzg' => row['betrag'].negative? ? 'H' : 'S', 'dmbtr' => format('%.2f', row['betrag'].abs)) + end + select %w[belnr buzei datum konto shkzg dmbtr betrag buchungstext] + + # 3. abgleichen, bevor etwas geschrieben wird + totals 'betrag' + reconcile do |stats| + quelle, ergebnis = stats.total('betrag', :read), stats.total('betrag') + next "Summe geändert: Quelle #{DataPipeline.decimal(quelle)}, Ergebnis #{DataPipeline.decimal(ergebnis)}" if quelle != ergebnis + + "Soll und Haben ungleich, Saldo #{DataPipeline.decimal(ergebnis)}" unless ergebnis.zero? + end + + # 4. schreiben: Upsert macht den Lauf wiederholbar; die CSV-Datei im Zeichensatz des Zielsystems + write connection(:buchhaltung) + write container(:csv, file: "#{out}/buchungen_sap.csv", encoding: 'Windows-1252', csv: { col_sep: ';' }, + columns: %w[belnr buzei datum konto shkzg dmbtr buchungstext]) + + on_finish do |stats| + puts "#{stats.name}: #{stats}" + slack&.send_message("*#{stats.name}*: #{stats}") unless stats.ok? + end +end.run +``` + +| Teil | Wozu | +| --- | --- | +| `connections_file`, `connection(:buchhaltung)` | Zieldatenbank, Tabelle und Upsert-Schlüssel stehen in der Verbindungsdatei (6.2); in der Praxis genügt `config/connections.yml` ohne `connections_file` | +| `lock` | ein zweiter Export zur selben Zeit bricht sofort ab (6.3) | +| `history` | jeder Lauf hinterlässt eine JSON-Zeile, auch ein Abbruch (6.6) | +| `validate` ohne `on_error` | eine falsche Zeile bricht den Lauf ab, bevor etwas geschrieben ist | +| `default`, `rename`, `cast`, `select` | Kurzschritte statt kleiner `map`-Blöcke (3.4) | +| `totals`, `reconcile` | Summen werden exakt gerechnet; eine Abweichung bricht vor dem Schreiben ab (6.4) | +| `upsert: [belnr, buzei]` | ein wiederholter Lauf aktualisiert dieselben Zeilen, statt sie doppelt anzulegen (4.2) | +| `encoding: 'Windows-1252'` | Zeichensatz der Zieldatei (4.4) | +| `slack&.send_message` | Meldung nur bei Problemen und nur, wenn `SLACK_TOKEN` gesetzt ist (6.8) | + +```text +INFO start +INFO finish read=7 written=7 rejected=0 betrag=0/0 0.007s +buchungen_export: read=7 written=7 rejected=0 betrag=0/0 0.007s +``` + +`betrag=0/0` ist die Summe in der Quelle und im Ergebnis. Ein zweiter Lauf ergibt wieder 7 Zeilen in der Tabelle +`buchungen`; `tmp/runs.ndjson` hat dann zwei Einträge. Ändert man in `buchungen.csv` einen Betrag, bricht der Lauf +mit `ReconcileError: Soll und Haben ungleich, Saldo …` ab und schreibt nichts. + +Für Cron gehören Zeitraum, Sperre und Historie auf die Kommandozeile, damit sie für alle Pipelines eines Skripts +gelten (6.1): + +```sh +# crontab: täglich 06:00; Exit-Code 1 = Abbruch, 2 = verworfene Zeilen, 3 = lief noch +0 6 * * * cd /srv/etl && bundle exec etl run export.rb --from=$(date -v-1d +\%F) --lock=data/.export.lock --history=data/runs.ndjson >> log/export.log 2>&1 +``` + +(`date -v-1d` ist die macOS-Form; unter Linux `date -d yesterday +\%F`.) --- -## 2. Technische Dokumentation +## 3. Pipelines schreiben -### 2.1 Aufbau +### 3.1 Ablauf eines Laufs ```mermaid flowchart LR - subgraph Pipeline-Skript - DSL[DataPipeline.new do ... end] - end - DSL -->|container :typ| REG[Registry / autoload
<Typ>Storage] - REG --> SP[DataPipeline::StoragePlugin
read · write · write_row · query · close] - SP --- PL[postgres · sqlite · csv · json
excel · rest · eigene] - DSL -->|lookup| LK[(Nachschlage-Hash)] - SRC((Quelle)) --> STEPS[Schritte
validate · filter · map · transform] - LK -.-> STEPS - STEPS --> EDGE[header · trailer
ctx.read · ctx.written] + SRC((Quelle)) --> STEPS[Schritte
validate · filter · map
rename · select · cast · default] + LK[(lookup)] -.-> STEPS + STEPS --> REC{reconcile} + REC --> EDGE[header · trailer] EDGE --> DST1((Ziel 1)) EDGE -->|only:| DST2((Ziel 2)) STEPS -. on_error .-> REJ((Fehlerdatei)) - DSL -->|run| ST[Stats · on_finish
Exit-Code über bin/etl] + EDGE --> ST[Stats · history · on_finish] ``` -| Datei | Inhalt | -| --- | --- | -| [lib/datapipeline.rb](lib/datapipeline.rb) | DSL und Ausführung (`run`), `Stats`, `Context` | -| [lib/datapipeline/validation.rb](lib/datapipeline/validation.rb) | Regeln für `validate` | -| [lib/datapipeline/watermark.rb](lib/datapipeline/watermark.rb) | Wasserzeichen für inkrementelle Läufe | -| [lib/datapipeline/lock.rb](lib/datapipeline/lock.rb) | Sperre gegen parallele Läufe | -| [lib/datapipeline/connections.rb](lib/datapipeline/connections.rb) | Verbindungsdatei `config/connections.yml` | -| [lib/datapipeline/layout.rb](lib/datapipeline/layout.rb) | Satzarten-Layouts (Feldliste je Satzart, siehe 2.9) | -| [exe/etl](exe/etl) | Kommando `etl` des Gems mit Exit-Codes (siehe 2.7); [bin/etl](bin/etl) startet es im Repository mit dem Bundle | -| [datapipeline.gemspec](datapipeline.gemspec), [CHANGELOG.md](CHANGELOG.md) | Gem-Beschreibung, Versionsverlauf | -| [examples/](examples/) | Beispiele aus Abschnitt 1 mit Daten in `examples/data/`, Demo-API für die REST-Beispiele | -| [lib/datapipeline/storage/storage_plugin.rb](lib/datapipeline/storage/storage_plugin.rb) | Basisklasse der Storage-Plugins | -| [lib/datapipeline/storage/postgres.rb](lib/datapipeline/storage/postgres.rb) | Quelle/Ziel PostgreSQL (`pg`) | -| [lib/datapipeline/storage/csv.rb](lib/datapipeline/storage/csv.rb) | Quelle/Ziel CSV | -| [lib/datapipeline/storage/json.rb](lib/datapipeline/storage/json.rb) | Quelle/Ziel JSON und NDJSON | -| [lib/datapipeline/storage/sqlite.rb](lib/datapipeline/storage/sqlite.rb) | Quelle/Ziel SQLite (`sqlite3`) | -| [lib/datapipeline/storage/excel.rb](lib/datapipeline/storage/excel.rb) | Quelle/Ziel Excel xlsx (`roo`, `caxlsx`) | -| [lib/datapipeline/storage/rest.rb](lib/datapipeline/storage/rest.rb) | Quelle/Ziel REST-API (`http`) | -| [lib/datapipeline/storage/fixed.rb](lib/datapipeline/storage/fixed.rb) | Quelle/Ziel Festbreiten-Datei nach Layout | -| [lib/datapipeline/notification/slack.rb](lib/datapipeline/notification/slack.rb) | Benachrichtigung, z. B. aus `on_finish` | -| [lib/datapipeline/version.rb](lib/datapipeline/version.rb) | `DataPipeline::VERSION` | +1. `run` nimmt die Sperre (`lock`), falls gesetzt, und liest die Quelle: im Batch ganz in den Speicher, im + Streaming Zeile für Zeile. +2. Jede Quellzeile läuft der Reihe nach durch die Schritte. Ein Schritt gibt null, eine oder mehrere Zeilen weiter; + der nächste Schritt bekommt die Ausgabe des vorigen. +3. Scheitert ein Schritt für eine Zeile, entscheidet `on_error` (3.6). +4. Nach den Daten laufen `reconcile` (6.4) und dann `header` und `trailer`; sie kennen die Zähler. +5. Die Zeilen gehen an alle Ziele: im Batch auf einmal (`write`), im Streaming sofort je Zeile (`write_row`). +6. Am Ende schliesst `run` Quelle und Ziele (auch nach einem Fehler), speichert das Wasserzeichen (nur nach Erfolg), + gibt die Sperre frei, schreibt die Historie und ruft `on_finish` auf. -### 2.2 DSL-Referenz +### 3.2 DSL-Referenz + +Alle Aufrufe stehen im Block von `DataPipeline.new do … end`. Ein unbekannter Aufruf löst `NoMethodError` aus, +Tippfehler fallen also sofort auf. + +**Quelle und Ziele** | Aufruf | Wirkung | | --- | --- | -| `DataPipeline.settings(opts)` | Defaults für alle folgenden Pipelines; der Hash wird ersetzt, nicht verändert (laufende Pipelines behalten ihre Kopie); `reset_settings!` setzt zurück | -| `DataPipeline.register(:typ, Klasse \| 'Klassenname', path:)` | eigenes Plugin; mit `path:` erst bei der ersten Nutzung geladen | -| `DataPipeline.new(**settings) { ... }` | führt den Block per `instance_eval` aus; `settings` gelten nur für diese Pipeline, z. B. `new(limit: 10)` | -| `configure(opts)` | Einstellungen dieser Pipeline als Hash: `verbose`, `streamed`, `chained`, `version` | -| `version(str)`, `verbose(flag = true)`, `streamed(flag = true)`, `chained(flag = true)` | dasselbe als DSL-Aufruf | -| `container(:typ, **opts)` | erzeugt das registrierte Plugin, sonst `Storage.new(**opts)` (z. B. `:postgres` → `DataPipeline::PostgresStorage`); fehlt es: „Unsupported storage type“; fehlende Gems, Verbindungs- und Optionsfehler kommen unverändert durch | -| `read(storage, opts = {})` | legt die Quelle fest; `opts` gehen an `storage.set` | -| `transform(opts = {}) { \|row, index\| ... }` | **veraltet**, entfällt in 3.0 (siehe 2.3): Transformation; `processing: :prepend \| :normal \| :append`, `condition:` | -| `map(opts = {}) { \|row, ctx\| ... }` | Schritt in der Kette: Hash, Array oder `nil` zurückgeben; schaltet Verkettung ein | -| `filter(opts = {}) { \|row, ctx\| ... }` | Schritt in der Kette: Zeile bleibt, wenn der Block wahr ist; schaltet Verkettung ein | -| `rename 'kunnr' => 'kundennr'` | Kurzschritt: Schlüssel umbenennen, Position bleibt | -| `select %w[belnr betrag]` | Kurzschritt: nur diese Schlüssel in dieser Reihenfolge, fehlende als `nil` | -| `cast 'betrag' => :decimal` | Kurzschritt: umwandeln in `:integer`, `:decimal` (Float), `:date`, `:string`; leere Werte werden `nil`, ein Fehler läuft über `on_error` | -| `default 'land' => 'CH'` | Kurzschritt: Wert für `nil` oder leeren Text; ein Lambda bekommt die Zeile | -| `header { \|ctx\| ... }` | Zeile(n) vor den Daten; wird nach den Daten berechnet, kennt also `ctx.written` | -| `trailer { \|ctx\| ... }` | Zeile(n) nach den Daten | -| `query(storage, query: sql, params: [])` | führt sofort eine Abfrage aus (Subquery in einem Schritt) | +| `container(:typ, **opts)` | erzeugt ein Plugin (Kapitel 4); `connection: :name` nimmt die Optionen aus der Verbindungsdatei | +| `connection(:name, **opts)` | Plugin aus der Verbindungsdatei, Typ von dort (6.2) | +| `read(storage, opts = {})` | Quelle; `opts` gehen an `storage.set`, z. B. `read pg, query: '…', params: [...]` | +| `write(storage, only: ->(row) { … })` | Ziel; mehrfach möglich, mit `only:` nur die passenden Zeilen (auch Kopf- und Schlusssätze) | | `lookup(storage, key:, unique: false, query:, params:)` | lädt eine Tabelle einmal als Hash: `key => [Zeilen]`, mit `unique: true` `key => Zeile`; `key` als Spaltenname oder Lambda; ohne `query:` wird `storage.read` genutzt | -| `write(storage, only: ->(row) { ... })` | Ziel; mehrfach möglich, jedes Ziel bekommt alle Zeilen oder mit `only:` die passenden (auch Kopf- und Schlusssätze) | -| `validate(rules \| layout, warn: false, ignore: []) { \|row\| ... }` | Prüfschritt (siehe 2.7, 2.9); `warn: true` nur Warnung, Zeile bleibt; schaltet Verkettung ein | -| `on_error(:abort \| :skip \| :collect, to: storage)` | Umgang mit Fehlern in Datenschritten (siehe 2.7) | -| `on_finish { \|stats\| ... }` | nach jedem Lauf, auch nach Abbruch | -| `totals 'dmbtr', …` | Summen dieser Spalten über die Quellzeilen und über die Zeilen nach allen Schritten (siehe 2.7) | -| `reconcile { \|stats\| ... }` | Abgleich nach allen Schritten, vor dem Schreiben; eine Meldung bricht mit `DataPipeline::ReconcileError` ab | -| `name(str)`, `history(file)` | Name des Laufs; eine JSON-Zeile je Lauf in `file` (siehe 2.7) | -| `connection(:name, **opts)` | Storage aus der Verbindungsdatei (siehe 2.7); `container(:typ, connection: :name)` ebenso | -| `watermark(name, column:, file:, initial:)` | letzter verarbeiteter Wert für inkrementelle Läufe (siehe 2.7) | -| `lock(file)` | Sperre während `run`; läuft schon ein Lauf mit derselben Datei, bricht dieser mit `DataPipeline::LockError` ab (siehe 2.7) | -| `dry_run(flag = true)`, `limit(n)` | nichts schreiben bzw. nur die ersten n Quellzeilen | -| `dry_run?` | in Schritten mit Nebenwirkungen (API-Aufrufe): `true` im Dry-Run | -| `logger(logger)` | eigener Ruby-`Logger`; Default: Warnungen auf `$stderr` | -| `params` | Parameter von der Kommandozeile: über `bin/etl` oder direkt aus `ARGV` (`--name=wert`, `--param key=wert`) | +| `query(storage, query: sql, params: [])` | führt sofort eine Abfrage aus und liefert die Zeilen | + +**Schritte** (3.3, 3.4) + +| Aufruf | Wirkung | +| --- | --- | +| `map(opts = {}) { \|row, ctx\| … }` | Hash = eine Zeile, Array = mehrere, `nil` oder `[]` = keine | +| `filter(opts = {}) { \|row, ctx\| … }` | Zeile bleibt, wenn der Block wahr ist | +| `rename 'alt' => 'neu'` | Schlüssel umbenennen, Position bleibt | +| `select %w[a b c]` | nur diese Schlüssel in dieser Reihenfolge, fehlende als `nil` | +| `cast 'betrag' => :decimal` | umwandeln in `:integer`, `:decimal`, `:date`, `:string`; leer wird `nil` | +| `default 'land' => 'CH'` | Wert für `nil` oder leeren Text; ein Lambda bekommt die Zeile | +| `validate(regeln \| layout, warn: false, ignore: []) { \|row\| … }` | Prüfschritt (3.5, Kapitel 5) | +| `header { \|ctx\| … }`, `trailer { \|ctx\| … }` | Zeile(n) vor bzw. nach den Daten | + +**Fehler und Ergebnis** (3.6, 3.7, Kapitel 6) + +| Aufruf | Wirkung | +| --- | --- | +| `on_error(:abort \| :skip \| :collect, to: storage)` | Umgang mit Fehlern in Schritten | +| `totals 'spalte', …` | Summen über Quelle und Ergebnis | +| `reconcile { \|stats\| … }` | Abgleich vor dem Schreiben; eine Meldung bricht ab | +| `on_finish { \|stats\| … }` | nach jedem Lauf, auch nach einem Abbruch | +| `name(str)`, `version(str)` | Name und Version des Laufs für Log, Stats und Historie | + +**Betrieb** (Kapitel 6) + +| Aufruf | Wirkung | +| --- | --- | +| `lock(file)` | Sperre gegen parallele Läufe | +| `history(file)` | eine JSON-Zeile je Lauf | +| `watermark(name, column:, file:, initial:)` | höchster verarbeiteter Wert für inkrementelle Läufe | +| `dry_run(flag = true)`, `dry_run?` | nichts schreiben; `dry_run?` für eigene Schritte mit Nebenwirkungen | +| `limit(n)` | nur die ersten n Quellzeilen | +| `params` | Parameter von der Kommandozeile (`--name=wert`, `--param key=wert`) | +| `logger(logger)` | eigener Ruby-`Logger`; Default: Warnungen auf `$stderr`, über `etl` Info | + +**Ausführung und Einstellungen** + +| Aufruf | Wirkung | +| --- | --- | +| `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` | -| jeder andere Aufruf | `NoMethodError` (Tippfehler fallen sofort auf) | +| `DataPipeline.new(**settings) { … }` | Einstellungen nur für diese Pipeline, z. B. `new(limit: 10)` | +| `DataPipeline.settings(opts)` | Defaults für alle folgenden Pipelines (so setzt `etl` Dry-Run und Log-Level); `reset_settings!` setzt zurück | +| `DataPipeline.register(:typ, Klasse \| 'Klasse', path:)` | eigenes Plugin bekannt machen (4.1) | +| `DataPipeline.params`, `DataPipeline.runs` | Parameter bzw. Stats aller Läufe des Prozesses (ausserhalb des Blocks) | +| `DataPipeline.history(file, name:)`, `DataPipeline.decimal(r)` | Historie lesen (6.6), Summe als Dezimaltext (6.4) | -### 2.3 Ausführungsmodell (`run`) +### 3.3 Schritte -> **Veraltet ab 2.1, entfällt in 3.0:** `transform`, der Parallelmodus (mehrere `transform` ohne `chained`) und -> `condition:` als String (`eval`). Ab 3.0 laufen alle Schritte verkettet. Wer sie nutzt, bekommt je Prozess einmal -> eine Warnung im Log (`WARN deprecated: …`). Umstellung: `transform` → `map` (Block bekommt `ctx` statt `index`, -> `ctx.index`), `processing: :prepend`/`:append` → `header`/`trailer`, `condition: "row['a'] == 1"` → -> `condition: ->(row) { row['a'] == 1 }`. Das bisherige Verhalten beschreibt der Rest dieses Abschnitts. +**Rückgabe** – Ein `map`-Block gibt einen Hash (eine Zeile), ein Array (mehrere Zeilen, z. B. Kopf + Positionen) +oder `nil`/`[]` (keine Zeile) zurück; `nil` innerhalb eines Arrays wird übersprungen. `filter` behält die Zeile, wenn +der Block wahr ist. -1. Batch (Default): `data = source.read` lädt die gesamte Quelle in den Speicher. Streaming: siehe unten. -2. Die Schritte werden in drei Gruppen geteilt, innerhalb jeder Gruppe gilt die Reihenfolge der Registrierung: - **vorne** (`:prepend`, `header`), **Daten** (`:normal`, `map`, `filter`), **hinten** (`:append`, `trailer`). -3. Die Daten-Schritte laufen in einem von zwei Modi: +**Kontext** – Der zweite Blockparameter `ctx` (`DataPipeline::Context`) enthält `index` (Position der Zeile in diesem +Schritt, ab 0), `read` (Anzahl Quellzeilen, im Streaming `nil`) und in `header`/`trailer` `written` (Anzahl +Datenzeilen nach allen Schritten). - | Modus | Wann | Eingabe eines Schritts | - | --- | --- | --- | - | parallel | Default | immer eine eigene Kopie der Quellzeilen; die Ergebnisse aller Schritte werden hintereinander angehängt | - | verkettet | `chained`, oder sobald `map`/`filter` vorkommt, oder ganz ohne Schritte | die Ausgabe des vorherigen Schritts | +```ruby +map { |row, ctx| row.merge('lfdnr' => format('%06d', ctx.index + 1)) } +trailer { |ctx| { 'satyp' => 'ENDE', 'anzahl' => ctx.written } } +``` - 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`. -4. Rückgabe eines Blocks: ein Hash ergibt eine Zeile, ein Array mehrere Zeilen (1:n, z. B. Kopf + Positionen), - `nil` oder `[]` keine Zeile; `nil` innerhalb eines Arrays wird übersprungen. -5. `condition:` an `transform` oder `map`: String (per `eval`, Variable `row` sichtbar) oder Lambda `->(row) { ... }`. - Zeilen, die die Bedingung nicht erfüllen, fallen in diesem Schritt weg. -6. Vorne und hinten: +**Bedingung** – `condition: ->(row) { … }` an `map` oder `filter` wendet den Schritt nur auf passende Zeilen an; die +anderen fallen in diesem Schritt weg. - | Schritt | Aufruf | Zeitpunkt | - | --- | --- | --- | - | `transform(processing: :prepend)` | `(nil, 0)` | vor den Daten | - | `header` | `(ctx)` mit `read`, `written` | nach den Daten, steht aber vorne | - | `transform(processing: :append)` | `(nil, n)`: parallel n = Quellzeilen, verkettet n = Datenzeilen | nach den Daten | - | `trailer` | `(ctx)` mit `read`, `written` | nach den Daten | +**Kopf- und Schlusssätze** – `header` steht vorne in der Ausgabe, wird im Batch aber erst nach den Daten berechnet und +kennt deshalb `ctx.read` und `ctx.written`. Beide können eine Zeile oder ein Array zurückgeben. Ein Fehler in +`header` oder `trailer` bricht immer ab. -7. Ohne Schritte wird die Quelle unverändert ins Ziel kopiert. -8. Batch: `target.write(ausgabe)` schreibt alles auf einmal. +**Ohne Schritte** wird die Quelle unverändert in die Ziele kopiert. -Der Kontext `ctx` (`DataPipeline::Context`) enthält `read` (Anzahl Quellzeilen), `written` (Anzahl Datenzeilen nach allen -Schritten) und bei `map`/`filter` `index` (Position der Zeile im aktuellen Schritt). - -Mit `verbose` werden im Batch Quelle, Daten nach allen Schritten und Ziel als JSON ausgegeben. - -Nach dem Lauf schliesst `run` Quelle und Ziel (`close`), auch im Fehlerfall. - -**Streaming-Modus** (`streamed`): Jede Zeile geht einzeln von `source.read { |row| ... }` durch die Kette und sofort -an `target.write_row`; am Ende `target.finalize`. Der Speicherbedarf bleibt konstant (getestet: 500 000 CSV-Zeilen -mit 30 MB statt 614 MB im Batch). Die Datenzeilen sind dieselben wie im Batch, mit diesen Unterschieden: +**Batch und Streaming** – Im Batch (Default) liest `source.read` alles in den Speicher, am Ende schreibt +`target.write` alles auf einmal. Mit `streamed` geht jede Zeile einzeln von `source.read { |row| … }` durch die +Schritte und sofort an `target.write_row`; am Ende ruft `run` `target.finalize` auf. Der Speicherbedarf bleibt +konstant (gemessen: 500 000 CSV-Zeilen mit 30 MB statt 614 MB im Batch). Die Datenzeilen sind dieselben, mit diesen +Unterschieden: | Thema | Batch | Streaming | | --- | --- | --- | | `header` | kennt `read` und `written` | wird zuerst geschrieben, `read` und `written` sind `nil` | | `ctx.read` in `map`/`filter` | Anzahl Quellzeilen | `nil` | -| mehrere `transform` im Parallelmodus | erlaubt | `ArgumentError`, dafür `chained` setzen | -| Postgres als Ziel | eine Transaktion für alles | eine Transaktion je `batch_size` Zeilen (Default 1000); `batch_size: :all` eine für alles | +| `reconcile` | vor dem Schreiben | nach der letzten Zeile; die Zeilen sind schon geschrieben | +| Postgres als Ziel | eine Transaktion für alles | eine Transaktion je `batch_size` Zeilen, mit `batch_size: :all` eine für alles (4.2) | +| Postgres als Quelle | eine Abfrage | serverseitiger Cursor, je `fetch_size` Zeilen | -Postgres als Quelle liest im Streaming über einen serverseitigen Cursor (`fetch_size`, Default 1000). Abfragen und -Lookups auf derselben Verbindung bleiben währenddessen möglich. +`limit n` beendet im Streaming das Lesen nach n Zeilen. -### 2.4 Plugin-Schnittstelle +### 3.4 Kurzschritte -Ein Storage-Plugin ist eine Klasse `Storage < DataPipeline::StoragePlugin` mit: +Häufige kleine `map`-Blöcke gibt es als eigene Schritte. Sie verhalten sich wie `map` und heissen in Log und +Fehlerdatei nach sich selbst (`cast#3`). + +| Schritt | Wirkung | Beispiel | +| --- | --- | --- | +| `rename` | Schlüssel umbenennen; die Position in der Zeile bleibt | `rename 'kunnr' => 'kundennr', 'name1' => 'name'` | +| `select` | nur diese Schlüssel, in dieser Reihenfolge; fehlende werden `nil` | `select %w[kundennr name land]` | +| `cast` | Wert umwandeln, leere Werte (`nil`, `''`, nur Leerzeichen) werden `nil` | `cast 'betrag' => :decimal, 'datum' => :date` | +| `default` | Wert für `nil` oder leeren Text; ein Lambda wird mit der Zeile aufgerufen | `default 'land' => 'CH', 'erfasst' => ->(row) { row['geaendert'] }` | + +| `cast`-Typ | Ergebnis | Gültig | +| --- | --- | --- | +| `:integer` | Integer | ganze Zahl zur Basis 10 (`'007'` → 7) | +| `:decimal` | Float | alles, was Rubys `Float()` annimmt (`'12.5'`, `'1e3'`) | +| `:date` | Date | `Date` oder ISO-Text `YYYY-MM-DD` | +| `:string` | String | jeder Wert | + +Kann ein Wert nicht umgewandelt werden, scheitert der Schritt für diese Zeile mit `betrag: "x" is not decimal` und +läuft über `on_error`. Ein unbekannter Typ fällt schon beim Aufbau der Pipeline auf. + +### 3.5 Validierung + +`validate` prüft Spalten nach Regeln, nach einem Layout (Kapitel 5) und/oder mit einem Block. Eine verletzte Regel +löst `DataPipeline::ValidationError` aus und läuft damit über `on_error`; alle Meldungen einer Zeile stehen, mit `; ` +getrennt, in einer Fehlermeldung. Leere Werte (`nil`, `''`, nur Leerzeichen) prüft nur `required`. + +| Regel | Prüft | Meldung | +| --- | --- | --- | +| `required: true` | Wert vorhanden und nicht leer | `konto: missing` | +| `max: 35`, `min: 1` | Länge als Text | `name1: longer than 35`, `… shorter than 1` | +| `format: /\A\d+\z/` | regulärer Ausdruck auf den Text | `konto: format /\A\d+\z/` | +| `in: %w[S H]` | erlaubte Werte | `shkzg: not in ["S", "H"]` | +| `type: :integer` | ganze Zahl zur Basis 10 | `menge: not integer` | +| `type: :decimal` | alles, was Rubys `Float()` annimmt | `dmbtr: not decimal` | +| `type: :date` | `Date` oder ISO-Text `YYYY-MM-DD` | `datum: not date` | + +```ruby +on_error :collect, to: container(:csv, file: 'data/fehler.csv', columns: %w[belnr konto dmbtr _step _error]) +validate 'konto' => { required: true, format: /\A\d+\z/ }, + 'shkzg' => { in: %w[S H] }, + 'dmbtr' => { type: :decimal }, + 'name1' => { max: 35 } +validate { |row| 'Betrag 0' if row['dmbtr'].to_f.zero? } # nil = in Ordnung, Text oder Array = Fehler +validate LAYOUT, warn: true, ignore: %w[BKTXT] # Layout; nur Warnung; bekannte Lücke auslassen +``` + +Mit `warn: true` bleibt die Zeile drin, es gibt eine Warnung im Log und `stats.warnings` zählt mit (ins Log kommen +höchstens 100). `ignore: %w[FELD]` lässt Meldungen zu diesen Feldern weg. `validate` prüft, wandelt aber nicht um; +dafür gibt es `cast` (3.4). Die Feldtypen eines Layouts und ihr Unterschied zu diesen Regeln stehen in 5.6. + +### 3.6 Fehler in Schritten + +`on_error` legt fest, was passiert, wenn ein Schritt (`map`, `filter`, `validate`, Kurzschritte) für eine Zeile eine +Exception wirft. Fehler beim Lesen, beim Schreiben, in `header`, `trailer` und `reconcile` brechen immer ab. + +| Modus | Wirkung | +| --- | --- | +| `:abort` (Default) | Lauf bricht ab, die Exception wird weitergereicht | +| `:skip` | Zeile fällt weg, Warnung im Log, Zähler `rejected` | +| `:collect, to: storage` | wie `:skip`, zusätzlich die Eingangszeile mit `_step` und `_error` per `write_row` ins Ziel `storage` | + +Ein Schritt heisst in Meldungen `#` in der Reihenfolge im Skript, z. B. `validate#1`, `cast#3`. Im +Dry-Run wird die Fehlerdatei nicht geschrieben. + +### 3.7 Ergebnis und Log + +`run` liefert `DataPipeline::Stats`; dasselbe Objekt bekommen `reconcile` und `on_finish`. + +| Feld | Inhalt | +| --- | --- | +| `name`, `version` | aus `name`/`version`; über `etl` ist `name` ohne Angabe der Skriptname ohne `.rb` | +| `read` | gelesene Quellzeilen | +| `written` | Datenzeilen nach allen Schritten (ohne Kopf- und Schlusssätze) | +| `output` | an die Ziele gegebene Zeilen inkl. Kopf- und Schlusssätzen | +| `rejected`, `errors` | verworfene Zeilen; die ersten 100 als Hash mit `step`, `error` und der Eingangszeile `row` | +| `warnings` | Zeilen mit Warnung aus `validate …, warn: true` | +| `totals`, `total(spalte, :read \| :written)` | Summen aus `totals` als `Rational` (6.4) | +| `started_at`, `finished_at`, `duration` | Zeitpunkte und Laufzeit in Sekunden | +| `dry_run` | Lauf war ein Dry-Run | +| `error` | Exception eines abgebrochenen Laufs | +| `ok?` | kein Abbruch und nichts verworfen | +| `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 +Abbruch; ein Fehler im Hook wird nur geloggt und ändert weder Ergebnis noch Exit-Code. + +```ruby +on_finish { |stats| warn "Buchungen: #{stats}" unless stats.ok? } +``` + +--- + +## 4. Plugins + +`container(:typ, **opts)` erzeugt ein Plugin. Die mitgelieferten liegen im Namensraum `DataPipeline` und werden mit +ihren Gems erst bei der ersten Nutzung geladen. Alle liefern Zeilen mit String-Schlüsseln und unterstützen `read`, +`read { |row| }`, `write(rows)`, `write_row(row)` und `close`. + +| `container` | Klasse | Gem | Abschnitt | +| --- | --- | --- | --- | +| `:postgres` | `DataPipeline::PostgresStorage` | `pg` | 4.2 | +| `:sqlite`, `:sqlite3` | `DataPipeline::SqliteStorage` | `sqlite3` | 4.3 | +| `:csv` | `DataPipeline::CsvStorage` | – | 4.4 | +| `:json` | `DataPipeline::JsonStorage` | – | 4.5 | +| `:excel`, `:xlsx` | `DataPipeline::ExcelStorage` | `roo` (lesen), `caxlsx` (schreiben) | 4.6 | +| `:fixed` | `DataPipeline::FixedStorage` | – | 4.7 | +| `:rest` | `DataPipeline::RestStorage` | `http` | 4.8 | + +### 4.1 Eigene Plugins + +Ein Plugin ist eine Klasse mit Basisklasse `DataPipeline::StoragePlugin`. Es braucht nur die Methoden, die es +anbietet; die Basisklasse meldet fehlende mit `NotImplementedError`. | Methode | Pflicht | Verwendung | | --- | --- | --- | -| `initialize(**opts, &block)` | ja | Verbindung/Datei vorbereiten | -| `read` → `Array>` | als Quelle | Batch-Lesen | -| `read { \|row\| }` | als Quelle | zeilenweises Lesen | -| `write(rows)` | als Ziel | Batch-Schreiben | -| `write_row(row)` | als Ziel | zeilenweises Schreiben | -| `query(query: sql, params: [])` → `Array` | optional | Ad-hoc-Abfragen aus Transformationen | -| `set(opts)` | Basisklasse | Optionen nachreichen, wird von `read(storage, opts)` genutzt | -| `close` / `finalize` | Basisklasse | Verbindung/Datei freigeben; `run` ruft `close` für Quelle und Ziel am Ende auf | - -Alle Plugins liefern Zeilen mit **String-Schlüsseln**. Die mitgelieferten Plugins liegen im Namensraum `DataPipeline` (`DataPipeline::CsvStorage` usw.) und werden per `autoload` erst bei der ersten Nutzung geladen. Neues Plugin: Klasse `Storage < DataPipeline::StoragePlugin` in der Anwendung schreiben und mit `DataPipeline.register :name, NameStorage` (oder `'NameStorage', path: 'datei.rb'` für Laden bei Bedarf) bekannt machen – danach ist `container :name` verfügbar. Ohne `register` findet `container :name` eine Klasse `Storage` auf oberster Ebene. - -### 2.5 Mitgelieferte Plugins - -**DataPipeline::PostgresStorage** – Optionen `host`, `port` (Default 5432), `database`, `username`, `password`, `query`, `params`, -`table`, `typed`, `fetch_size`, `truncate`, `upsert`, `batch_size`. -- `read` / `query` mit Platzhaltern `$1, $2, …` und `params: [...]`; Ergebnis als `Hash` - (`NULL` = `nil`), mit `typed: true` als Ruby-Typen (Integer, Float, Date, …). -- `read { |row| }` liest über einen Cursor in einer Transaktion, je `fetch_size` Zeilen (Default 1000). -- `write(rows)` schreibt in `table:` (oder alte Form `write(table, rows)`), alle Zeilen in **einer Transaktion**, - Werte als Bind-Parameter, `nil` = `NULL`, Tabellen-/Spaltennamen gequotet (`schema.tabelle` erlaubt). -- `truncate: true` leert die Tabelle vor dem ersten Schreiben (bei `write` in derselben Transaktion). -- `upsert: %w[belnr]` schreibt `insert … on conflict (belnr) do update set …` für alle übrigen Spalten der Zeile (ohne - übrige Spalten `do nothing`). Die Tabelle braucht einen Primärschlüssel oder Unique-Index auf genau diesen Spalten. - Damit kann ein Lauf nach einem Abbruch wiederholt werden, ohne doppelte Zeilen. -- `write_row` (Streaming) sammelt die Zeilen und schreibt je `batch_size` Zeilen (Default 1000) in einer Transaktion; - `finalize` und `close` schreiben den Rest. Mit `batch_size: :all` läuft der ganze Stream in **einer** Transaktion: - `finalize` am Ende eines erfolgreichen Laufs bestätigt sie, nach einem Abbruch rollt `close` sie zurück, auch ein - `truncate`. Dafür braucht das Ziel eine eigene Verbindung (nicht dieselbe wie die Quelle). +| `initialize(**opts, &block)` | ja | Verbindung oder Datei vorbereiten; möglichst erst bei der ersten Nutzung öffnen (Dry-Run) | +| `read` → `Array>` | als Quelle | Batch | +| `read { \|row\| }` | als Quelle | Streaming | +| `write(rows)` | als Ziel | Batch | +| `write_row(row)` | als Ziel | Streaming und Fehlerdatei | +| `query(query:, params: [])` → `Array` | optional | `query` und `lookup` | +| `set(opts)` | Basisklasse | Optionen nachreichen, genutzt von `read(storage, opts)` | +| `finalize` | Basisklasse | Ende eines Streams, ruft `close` auf | +| `close` | Basisklasse | Verbindung oder Datei freigeben; mehrfacher Aufruf muss gehen | ```ruby -pg = container :postgres, host: 'localhost', database: 'crm', username: 'etl', table: 'kunde_import' +DataPipeline.register :verzeichnis, VerzeichnisStorage # danach container(:verzeichnis, …) +DataPipeline.register :sftp, 'SftpStorage', path: 'lib/sftp_storage.rb' # erst bei der ersten Nutzung geladen +``` + +Ohne `register` findet `container :name` eine Klasse `Storage` im Namensraum `DataPipeline` oder auf oberster +Ebene. Ein vollständiges Beispiel steht in 2.4, Testhilfen in 8.2. Ein Plugin, das ein Gem braucht, lädt es mit +`self.class.require_gem 'name'`; fehlt das Gem, gibt es einen verständlichen Hinweis. + +### 4.2 PostgreSQL + +```ruby +pg = container :postgres, host: 'localhost', database: 'crm', username: 'etl', password: ENV['CRM_PASSWORD'] read pg, query: 'select * from kunde where geaendert >= $1', params: [Date.today - 7] - -# wiederholbar: vorhandene Belege werden aktualisiert, im Streaming alles oder nichts -write container(:postgres, **db, table: 'beleg', upsert: %w[belnr], batch_size: :all) +write container(:postgres, connection: :crm, table: 'beleg', upsert: %w[belnr], batch_size: :all) ``` -**DataPipeline::CsvStorage** – Optionen `file` (Pflicht), `csv:` (an Rubys `CSV`, gemergt mit den Defaults `headers: true`, -`col_sep: ','`), `columns`, `encoding`, `replace`, `bom`, `counter`, `verbose`. -- `columns: %w[a b c]` legt die Spaltenreihenfolge fest: fehlende Felder bleiben leer, String- und Symbol-Schlüssel - werden gefunden, die Kopfzeile kommt aus `columns`. Ohne `columns` gilt die Hash-Reihenfolge (nötig für Dateien mit - mehreren Satzarten wie CUDO). -- Lesen ohne Kopfzeile (`headers: false`): Schlüssel aus `columns`, sonst `"0"`, `"1"`, … -- Leere Daten: ohne `columns` leere Datei, mit `columns` nur die Kopfzeile. -- `layout: DataPipeline::Layout` schreibt Dateien mit mehreren Satzarten in der Feldreihenfolge je Satzart, ohne - Kopfzeile; ein Feld, das nicht im Layout steht, bricht ab (siehe 2.9). -- `encoding: 'ISO-8859-1'` (oder `'Windows-1252'`, …) ist der Zeichensatz der Datei, Default UTF-8. Gelesen wird als - UTF-8, geschrieben wird von UTF-8 umkodiert. Ein Zeichen, das der Zeichensatz nicht kennt (z. B. `€` in ISO-8859-1), - bricht mit `Encoding::UndefinedConversionError` ab; mit `replace: '?'` wird es ersetzt. Ein UTF-8-BOM am - Dateianfang wird beim Lesen übersprungen, `bom: true` schreibt eines (für Excel). Ein `encoding` in `csv:` wird wie - bisher unverändert an Ruby weitergegeben. +| Option | Bedeutung | +| --- | --- | +| `host`, `port` (5432), `database`, `username`, `password` | Verbindung; ein Verbindungsfehler meldet `Error connection to database: …` | +| `query`, `params` | Abfrage für `read` mit Platzhaltern `$1, $2, …` | +| `typed: true` | Werte als Ruby-Typen (Integer, Float, Date, …) statt als Text | +| `fetch_size` | Zeilen je Abruf beim Lesen im Streaming (Default 1000) | +| `table` | Zieltabelle, auch `schema.tabelle`; alte Form `write(table, rows)` geht weiter | +| `truncate: true` | Tabelle vor dem ersten Schreiben leeren (in derselben Transaktion) | +| `upsert: %w[belnr]` | `insert … on conflict (belnr) do update set …` für alle übrigen Spalten der Zeile | +| `batch_size` | Streaming: Zeilen je Transaktion (Default 1000) oder `:all` | + +- **Lesen:** Ohne `typed` kommen alle Werte als Text, `NULL` als `nil`. Im Streaming liest ein serverseitiger Cursor + in einer Transaktion; Abfragen und Lookups auf derselben Verbindung bleiben dabei möglich. +- **Schreiben:** Werte gehen als Bind-Parameter, Tabellen- und Spaltennamen werden gequotet, `nil` wird `NULL`. + `write` schreibt alle Zeilen in **einer** Transaktion. +- **Upsert:** Die Tabelle braucht einen Primärschlüssel oder Unique-Index auf genau den Upsert-Spalten. Hat die Zeile + keine weiteren Spalten, wird `do nothing` geschrieben. Damit ist ein Lauf nach einem Abbruch wiederholbar. +- **Streaming:** `write_row` sammelt Zeilen und schreibt je `batch_size` in einer Transaktion; `finalize` und `close` + schreiben den Rest. Mit `batch_size: :all` läuft der ganze Stream in einer Transaktion: `finalize` am Ende eines + erfolgreichen Laufs bestätigt sie, nach einem Abbruch rollt `close` sie zurück (auch ein `truncate`). Dafür braucht + das Ziel eine eigene Verbindung, nicht dieselbe wie die Quelle. + +### 4.3 SQLite ```ruby -write container(:csv, file: 'out/cudo.txt', encoding: 'Windows-1252', layout: LAYOUT, csv: { headers: false, col_sep: '|' }) +write container(:sqlite, database: 'tmp/kunden.sqlite3', table: 'kunden', create: true, truncate: true) ``` -Übersicht der übrigen Plugins (alle mit `read`, `read { }`, `write`, `write_row`, `close`): +| Option | Bedeutung | +| --- | --- | +| `database` | Datei oder `':memory:'`; wird erst bei der ersten Nutzung geöffnet, ein Dry-Run legt keine an | +| `query`, `params` | Abfrage für `read`; `params` als Array für `?`/`$1`, als Hash für `:name` | +| `table` | Zieltabelle | +| `create: true` | Tabelle beim ersten Schreiben aus der ersten Zeile anlegen (Integer → `integer`, Float → `real`, sonst `text`) | +| `truncate: true` | Tabelle vor dem ersten Schreiben leeren | +| `upsert: %w[belnr]` | wie bei Postgres; mit `create: true` wird der nötige Unique-Index mit angelegt | -| `container` | Klasse | Gem (bei Nutzung geladen) | Wichtige Optionen | -| --- | --- | --- | --- | -| `:json` | `DataPipeline::JsonStorage` | – | `file`, `format: :array \| :lines` (Default aus Endung: `.ndjson`/`.jsonl` → Zeilen), `root: 'data.items'`, `pretty`; ein einzelnes Objekt wird als eine Zeile gelesen | -| `:sqlite`, `:sqlite3` | `DataPipeline::SqliteStorage` | `sqlite3` | `database` (Datei oder `':memory:'`), `query`, `params` (Array für `?`, Hash für `:name`), `table`, `create`, `truncate`, `upsert` | -| `:excel`, `:xlsx` | `DataPipeline::ExcelStorage` | `roo` (lesen), `caxlsx` (schreiben) | `file`, `sheet` (Name/Index), `headers`, `columns`, `skip` (Titelzeilen) | -| `:fixed` | `DataPipeline::FixedStorage` | – | `file`, `layout` (Pflicht), `encoding`, `replace`, `line_end`; Festbreiten-Datei, siehe 2.9 | -| `:rest` | `DataPipeline::RestStorage` | `http` | `url`, `auth`, `headers`, `params`, `root`, Paging `page_param` oder `next: ->(body) { … }`, `retries`, `timeout`, `batch_size`, `write_method`, `write_url` (auch als Lambda je Zeile), `ssl_verify` | +Werte kommen mit SQLite-Typen zurück (Integer, Float, String, `nil`); `Date` und `Time` werden als ISO-Text +geschrieben. `write` schreibt in einer Transaktion. -**DataPipeline::JsonStorage** – NDJSON wird zeilenweise gelesen und geschrieben. Ein JSON-Array wird beim Streaming als -`[`…`]` geschrieben und in `close` abgeschlossen, die Datei ist also auch beim Streaming gültiges JSON. +### 4.4 CSV -**DataPipeline::SqliteStorage** – Verhält sich wie Postgres: Bind-Parameter, Schreiben in einer Transaktion, Tabellen- und -Spaltennamen gequotet, `nil` = `NULL`, Datum als ISO-Text. Werte kommen mit SQLite-Typen zurück (Integer, Float, String). -`create: true` legt die Tabelle beim ersten Schreiben aus der ersten Zeile an (Integer → `integer`, Float → `real`, -sonst `text`), `truncate: true` leert sie vorher. `upsert: %w[belnr]` wie bei Postgres; mit `create: true` wird der -nötige Unique-Index gleich mit angelegt. Die Datei wird erst bei der ersten Nutzung geöffnet; ein Dry-Run -legt keine an. +```ruby +read container(:csv, file: 'kunden.csv', csv: { col_sep: ';' }) +write container(:csv, file: 'export.csv', columns: %w[id name land], encoding: 'Windows-1252') +``` -**DataPipeline::ExcelStorage** – Lesen von xlsx zeilenweise (`each_row_streaming`), leere Zeilen werden übersprungen. Beim -Schreiben bleibt Text immer Text: `'0000000001'` behält die Nullen, `'=1+1'` wird keine Formel. Zahlen und Datumswerte -behalten ihren Typ. Die Datei entsteht in `close` (caxlsx baut sie im Speicher auf). +| Option | Bedeutung | +| --- | --- | +| `file` | Pfad, Pflicht | +| `csv:` | Optionen für Rubys `CSV`, gemergt mit den Defaults `headers: true`, `col_sep: ','` | +| `columns` | Spaltenreihenfolge beim Schreiben (fehlende Felder bleiben leer, Kopfzeile aus `columns`); Schlüssel beim Lesen ohne Kopfzeile | +| `layout` | Satzarten-Layout: schreiben in Feldreihenfolge ohne Kopfzeile, lesen mit `headers: false` (5.4) | +| `encoding` | Zeichensatz der Datei, Default UTF-8 | +| `replace` | Ersatz für Zeichen, die der Zeichensatz nicht kennt, z. B. `'?'` | +| `bom: true` | UTF-8-BOM schreiben (für Excel) | +| `counter`, `verbose` | einen Punkt je geschriebene Zeile bzw. die Optionen ausgeben | -**DataPipeline::RestStorage** – Liest JSON, der Pfad zum Array kommt aus `root`. Paging über eine Seitennummer (bis eine Seite -leer ist) oder über eine Folge-URL aus der Antwort; `max_pages` (Default 1000) stoppt endlose Schleifen. Bei 429, 5xx -und Netzwerkfehlern wird `retries`-mal wiederholt (Wartezeit `retry_wait` × Versuch), 4xx bricht sofort mit -`DataPipeline::RestStorage::Error` ab. Schreiben je Zeile oder mit `batch_size` als JSON-Array; `write_url: ->(row) { … }` baut -je Zeile eine eigene URL (z. B. `PUT /kunden/`, nicht zusammen mit `batch_size`). `query(query: 'pfad', params:)` -hängt den Pfad an `url` an und erlaubt `lookup` gegen die API. Ein Fehler beim Schreiben bricht den Lauf ab (siehe 1.4). +- Ohne `columns` gilt beim Schreiben die Reihenfolge des Hashes; Kopfzeile aus den Schlüsseln der ersten Zeile. +- Lesen ohne Kopfzeile (`headers: false`): Schlüssel aus `columns`, sonst `"0"`, `"1"`, … +- Leere Daten: ohne `columns` eine leere Datei, mit `columns` nur die Kopfzeile. Im Dry-Run entsteht keine Datei. +- **Zeichensatz:** Gelesen wird immer als UTF-8, geschrieben wird von UTF-8 umkodiert. Ein Zeichen, das der + Zeichensatz nicht kennt (z. B. `€` in ISO-8859-1; Windows-1252 kennt es), bricht mit + `Encoding::UndefinedConversionError` ab, mit `replace: '?'` wird es ersetzt. Ein UTF-8-BOM am Dateianfang wird beim + Lesen übersprungen. Ein `encoding` in `csv:` wird wie früher unverändert an Ruby weitergegeben. -Für APIs im RPC-Stil (Suchen und Anlegen per POST mit Body, z. B. die pds API) gibt es einzelne Aufrufe: -`call(:post, 'person/listkunden', json: { … })` liefert das geparste JSON. Mit `retries: 0` wird ein anlegender Aufruf -bei `5xx` nicht wiederholt. Ein Fehler wirft `DataPipeline::RestStorage::Error` mit `status` (HTTP-Code, `nil` bei Netzwerkfehlern). -`ssl_verify: false` akzeptiert jedes Server-Zertifikat (eigene CA, Zugriff über IP-Adresse). +### 4.5 JSON und NDJSON + +```ruby +read container(:json, file: 'antwort.json', root: 'data.items') +write container(:json, file: 'bewegungen.ndjson') +``` + +| Option | Bedeutung | +| --- | --- | +| `file` | Pfad | +| `format` | `:array` oder `:lines`; Default aus der Endung (`.ndjson`, `.jsonl` → Zeilen) | +| `root` | Pfad zum Array beim Lesen, z. B. `'data.items'` | +| `pretty` | eingerücktes JSON | + +Ein einzelnes Objekt wird als eine Zeile gelesen. NDJSON wird zeilenweise gelesen und geschrieben; ein JSON-Array wird +im Streaming als `[`…`]` geschrieben und in `close` abgeschlossen, die Datei ist also immer gültiges JSON. + +### 4.6 Excel + +```ruby +read container(:excel, file: 'kunden.xlsx', sheet: 'Kunden', skip: 2) +write container(:excel, file: 'eu_kunden.xlsx', sheet: 'EU-Kunden', columns: %w[kundennr name land]) +``` + +Optionen: `file`, `sheet` (Name oder Index), `headers`, `columns`, `skip` (Titelzeilen vor der Kopfzeile). Gelesen +wird zeilenweise, leere Zeilen werden übersprungen. Beim Schreiben bleibt Text immer Text: `'0000000001'` behält die +Nullen, `'=1+1'` wird keine Formel; Zahlen und Datumswerte behalten ihren Typ. Die Datei entsteht in `close`. + +### 4.7 Festbreiten-Dateien + +```ruby +write container(:fixed, file: 'out/cudo.txt', layout: LAYOUT, encoding: 'Windows-1252') +read container(:fixed, file: 'in/rueckmeldung.txt', layout: LAYOUT) +``` + +Dateien ohne Trennzeichen, in denen jedes Feld eine feste Anzahl Zeichen hat. Die Breiten kommen aus einem +Satzarten-Layout; Optionen `file`, `layout` (beide Pflicht), `encoding`, `replace` (wie CSV) und `line_end` +(`"\n"` oder `"\r\n"`). Breiten, Ausrichtung und Lesen sind in 5.5 beschrieben. + +### 4.8 REST-APIs ```ruby api = container :rest, url: 'https://api.example.com/v1/kunden', auth: "Bearer #{ENV['API_TOKEN']}", - root: 'data', page_param: 'page' + root: 'data', page_param: 'page', retries: 3 +``` + +| Option | Bedeutung | +| --- | --- | +| `url`, `auth`, `headers`, `params` | Adresse, `Authorization`-Header, weitere Header, Query-Parameter | +| `root` | Pfad zum Array in der Antwort, z. B. `'data'` | +| `page_param` | Paging über eine Seitennummer, bis eine Seite leer ist | +| `next: ->(body) { … }` | Paging über eine Folge-URL aus der Antwort (`nil` = Ende) | +| `max_pages` | Schutz vor endlosen Schleifen (Default 1000) | +| `retries`, `retry_wait`, `timeout` | Wiederholung bei 429, 5xx und Netzwerkfehlern (Wartezeit `retry_wait` × Versuch) | +| `batch_size` | Schreiben in Paketen als JSON-Array | +| `write_method`, `write_url` | HTTP-Methode beim Schreiben; `write_url: ->(row) { … }` eine URL je Zeile (nicht mit `batch_size`) | +| `ssl_verify: false` | jedes Server-Zertifikat akzeptieren (eigene CA, Zugriff über IP-Adresse) | + +- `4xx` bricht sofort mit `DataPipeline::RestStorage::Error` ab (`status` enthält den HTTP-Code, `nil` bei + Netzwerkfehlern). Ein Fehler beim Schreiben bricht den Lauf ab. +- `query(query: 'pfad', params:)` hängt den Pfad an `url` an und erlaubt `lookup` gegen die API (2.5). +- Für APIs im RPC-Stil (Suchen und Anlegen per POST mit Body, z. B. die pds API) gibt es einzelne Aufrufe: + `call(:post, 'person/listkunden', json: { … })` liefert das geparste JSON. Mit `retries: 0` wird ein anlegender + Aufruf bei `5xx` nicht wiederholt. + +### 4.9 Weitere Beispiele + +```ruby +# REST nach Excel +api = container :rest, url: 'https://api.example.com/v1/kunden', auth: "Bearer #{ENV['API_TOKEN']}", + root: 'data', page_param: 'page' xlsx = container :excel, file: 'data/kunden.xlsx', sheet: 'Kunden', columns: %w[id name land] read api -map { |r| r.slice('id', 'name', 'land') } +select %w[id name land] write xlsx ``` -### 2.6 Abhängigkeiten +### 4.10 Abhängigkeiten Das Gem selbst braucht nur `csv` und `logger` (Ruby `>= 3.4`). Die Gems der Plugins sind optional und gehören ins Gemfile der Anwendung, die das Plugin nutzt: @@ -805,86 +1132,300 @@ Gemfile der Anwendung, die das Plugin nutzt: | `:sqlite` | `sqlite3` | | `:excel` | `roo` (lesen), `caxlsx` (schreiben) | | `:rest`, `DataPipeline::Slack` | `http` | -| `:csv`, `:json` | – | +| `:csv`, `:json`, `:fixed` | – | -Im Repository bindet das `Gemfile` das Gem über `gemspec` ein und ergänzt die Gruppe `plugins` (`pg`, `sqlite3`, -`roo`, `caxlsx`, `http`) für die Plugin-Tests und die Gruppe `test` (`minitest`, `rake`). `Gemfile.lock` ist versioniert. -Gem bauen: `gem build datapipeline.gemspec`. +Eine CSV- oder JSON-Pipeline lädt keines dieser Gems. Fehlt ein Gem, meldet das Plugin bei der ersten Nutzung, welches +ins Gemfile gehört. -Plugins und ihre Gems werden erst bei der ersten Nutzung geladen (`autoload` und `DataPipeline::StoragePlugin.require_gem`): eine -CSV- oder JSON-Pipeline lädt keines dieser Gems. Wer ein Plugin nicht braucht, lässt die Gruppe weg -(`bundle config set without plugins`); bei Nutzung meldet das Plugin dann, welches Gem fehlt. +--- -### 2.7 Betrieb +## 5. Satzarten-Layouts -**Fehler je Zeile** – `on_error` legt fest, was passiert, wenn ein Datenschritt (`transform`, `map`, `filter`, -`validate`) für eine Zeile eine Exception wirft. Fehler in `header`, `trailer`, prepend und append brechen immer ab. +### 5.1 Wozu -| Modus | Wirkung | +Dateien für Schnittstellen wie SAP CUDO enthalten mehrere Satzarten (Datei-, Kopf-, Positions-, Schlusssatz) mit je +einer festen Feldliste. `DataPipeline::Layout` beschreibt diese Listen einmal; damit lassen sich Zeilen prüfen, in der +richtigen Feldreihenfolge schreiben (CSV oder feste Breite) und wieder lesen. Eine falsch benannte Spalte verschiebt +so nicht mehr still alle folgenden Felder, sondern bricht ab. Ein vollständiges Beispiel sind die CUDO-Satzarten im +Projekt `etl` (`cudo/layout.rb`), erzeugt aus der Schnittstellenbeschreibung. + +```ruby +LAYOUT = DataPipeline::Layout.new do + record 'DOHD', [['SATYP', 'AN', 4, true], ['LFDNR', 'N', 10, true], ['BKTXT', 'AN', 25, true], ...] + record 'LARC', [['SATYP', 'AN', 4, true], ['SUMME', 'N', 10, true]] +end + +validate LAYOUT, warn: true # Pflicht, Länge, Format je Feld +write container(:csv, file: 'x.ok', layout: LAYOUT, csv: { headers: false, col_sep: '|' }) +``` + +Die Satzart einer Zeile ist ihr erster Wert (oder der Wert von `type_field:`). Schlüssel werden ohne Rücksicht auf +Gross-/Kleinschreibung zugeordnet, als String oder Symbol. + +### 5.2 Felder und Datentypen + +Jedes Feld ist ein Array `[Name, Typ, Länge, Pflicht]`: + +| Teil | Bedeutung | | --- | --- | -| `:abort` (Default) | Lauf bricht ab, Exception wird weitergereicht | -| `:skip` | Zeile fällt weg, Warnung im Log, Zähler `rejected` | -| `:collect, to: storage` | wie `:skip`, zusätzlich die Eingangszeile mit `_step` und `_error` per `write_row` ins Ziel `storage` | +| Name | Spaltenname in der Zeile | +| Typ | `AN`, `N`, `DAT` oder `TIM`, Gross-/Kleinschreibung egal; fehlt er (`nil`), gilt `AN`; ein unbekannter Typ wird wie `AN` behandelt | +| Länge | Zahl oder bei Nachkommastellen ein String (`'11.2'`); fehlt sie, wird die Länge nicht geprüft | +| Pflicht | `true` = darf nicht leer sein; fehlt sie, ist das Feld optional | -**Kurzschritte** – Häufige kleine `map`-Blöcke gibt es als eigene Schritte. Sie schalten wie `map` die Verkettung ein -und heissen in Fehlermeldungen und der Fehlerdatei nach sich selbst (`cast#3`). +Geprüft wird der Wert als Text (`to_s`). Ein leerer Wert (`nil`, `''`, nur Leerzeichen) wird nur auf Pflicht geprüft. + +| Typ | Inhalt | Gültig | Länge | Meldung | +| --- | --- | --- | --- | --- | +| `AN` | Text | jeder Wert | höchstens `Länge` Zeichen | `NAME: longer than 10` | +| `N` | Zahl | Ziffern, optional Vorzeichen und Punkt als Dezimaltrenner | `10` = höchstens 10 Stellen ohne Nachkommastellen; `'11.2'` = höchstens 9 Vor- und 2 Nachkommastellen; Vorzeichen zählt nicht, führende Nullen schon | `NAME: not a number`, `… more than 9 digits`, `… more than 2 decimals` | +| `DAT` | Datum | genau 8 Ziffern `YYYYMMDD`, gültiges Kalenderdatum | – | `NAME: not a date YYYYMMDD` | +| `TIM` | Uhrzeit | genau 6 Ziffern `HHMMSS`, 00:00:00 bis 23:59:59 | – | `NAME: not a time HHMMSS` | + +### 5.3 Formatieren + +Ohne Option prüft das Layout nur und schreibt die Werte unverändert; ein `Date` würde als `2026-10-09` geprüft und +abgelehnt. Mit `Layout.new(format: true)` bringt das Layout die Werte vor dem Prüfen und Schreiben in das Textformat +ihres Typs. Geprüft wird dann genau der Text, der in die Datei kommt. + +| Typ | Eingabe | Ergebnis | +| --- | --- | --- | +| `N` | Integer, Float, Rational, BigDecimal | Nachkommastellen aus der Länge (`'11.2'` → 2), kaufmännisch gerundet: `12.345` → `12.35`, `-0.005` → `-0.01`, `7` → `7.00`; ein Float zählt mit seinem kürzesten Text, `12.345` also nicht als `12.3449999…` | +| `N` | Zahl als Text | auf die Nachkommastellen aufgefüllt (`'3.5'` → `3.50`); zu viele Nachkommastellen bleiben stehen und werden gemeldet, Text wird nie gerundet | +| `DAT` | `Date`, `Time`, `'2026-10-09'` | `20261009` | +| `TIM` | `Time`, `'07:05:03'` | `070503` | +| alle | `nil` | bleibt leer | + +Anderes bleibt unverändert und wird wie gewohnt geprüft. Führende Nullen ergänzt nur die Festbreiten-Datei (5.5); in +CSV braucht es dafür ein `map` (`format('%06d', …)`). ```ruby -rename 'kunnr' => 'kundennr', 'name1' => 'name' -default 'land' => 'CH', 'erfasst' => ->(row) { row['geaendert'] } -cast 'betrag' => :decimal, 'datum' => :date, 'menge' => :integer -select %w[kundennr name land betrag datum] +LAYOUT = DataPipeline::Layout.new(format: true) do + record 'KOPF', [['SATYP', 'AN', 4, true], ['DATUM', 'DAT', 8, true], ['SUMME', 'N', '11.2', true]] +end +LAYOUT.values({ 'satyp' => 'KOPF', 'datum' => Date.new(2026, 10, 9), 'summe' => 1234.5 }) +# => ["KOPF", "20261009", "1234.50"] ``` -**Validierung** – `validate` prüft Spalten nach Regeln, nach einem Layout (2.9) und/oder mit einem Block. Eine -verletzte Regel löst `DataPipeline::ValidationError` aus und läuft damit über `on_error`. Leere Werte prüft nur -`required`. Regeln je Spalte: `required: true`, `max:`/`min:` (Länge als Text), `format:` (Regex auf den Text), -`in:` (erlaubte Werte) und `type:` – `:integer` (ganze Zahl, Basis 10), `:decimal` (alles, was Rubys `Float()` -annimmt) oder `:date` (`Date` oder ISO-Text `YYYY-MM-DD`). Die Feldtypen eines Layouts (`AN`, `N`, `DAT`, `TIM`) und -ihr Unterschied zu diesen Regeln stehen in 2.9. Mit `warn: true` bleibt die Zeile drin, es gibt eine Warnung und `stats.warnings` zählt mit (ins Log -kommen höchstens 100); `ignore: %w[FELD]` lässt bekannte Lücken aus. +### 5.4 Schreiben und Lesen als CSV + +Mit `layout:` schreibt das CSV-Plugin jede Zeile in der Feldreihenfolge ihrer Satzart, ohne Kopfzeile. Fehlende Felder +bleiben leer; ein Feld, das nicht im Layout steht, bricht mit `DataPipeline::LayoutError` ab (ausser mit +`strict: false`, dann wird es weggelassen). + +Mit `headers: false` liest dasselbe Plugin eine solche Datei wieder: Jede Zeile bekommt die Feldnamen ihrer Satzart +(klein geschrieben) als Schlüssel. Eine unbekannte Satzart oder mehr Werte als Felder brechen mit `LayoutError` ab. ```ruby -on_error :collect, to: container(:csv, file: 'data/fehler.csv', columns: %w[belnr konto dmbtr _step _error]) -validate 'konto' => { required: true, format: /\A\d+\z/ }, - 'shkzg' => { in: %w[S H] }, - 'dmbtr' => { type: :decimal }, - 'name1' => { max: 35 } -validate { |row| 'Betrag 0' if row['dmbtr'].to_f.zero? } +read container(:csv, file: 'in/cudo.ok', layout: LAYOUT, csv: { headers: false, col_sep: '|' }) +# { 'satyp' => 'LARC', 'summe' => '12' } ``` -**Ergebnis, Log, Hooks** – `run` liefert `DataPipeline::Stats`: +### 5.5 Festbreiten-Dateien -| Feld | Inhalt | +`container(:fixed, file:, layout:)` schreibt und liest Dateien, in denen jedes Feld eine feste Anzahl Zeichen hat. +Jedes `AN`- und `N`-Feld braucht dafür eine Länge, sonst bricht schon `container` mit `LayoutError` ab und nennt die +Felder. + +| Typ | Breite | Ausrichtung | +| --- | --- | --- | +| `AN` | Länge | links, rechts mit Leerzeichen aufgefüllt | +| `N` | Vor- + Nachkommastellen + Punkt: `'11.2'` → 12, `5` → 5 | rechts, links mit Nullen aufgefüllt, Vorzeichen vorne: `-0000012.50` | +| `DAT`, `TIM` | 8 bzw. 6 | genau so viele Ziffern | +| leeres Feld | Breite des Felds | nur Leerzeichen | + +Beim Schreiben werden die Werte immer formatiert (wie `format: true`). Ein Wert, der länger als sein Feld ist, wird +nicht abgeschnitten, sondern bricht mit `LayoutError` ab; mit `validate LAYOUT` vorher fällt er schon als Zeilenfehler +auf und kann über `on_error` gesammelt werden. Beim Lesen erkennt das Plugin die Satzart an ihrem Feld; die Schlüssel +sind die Feldnamen klein geschrieben, die Werte ohne Auffüllung (Zahlen behalten ihre führenden Nullen, `cast` macht +daraus Zahlen), leere Felder sind `nil` und leere Zeilen werden übersprungen. Breiten zählen Zeichen, nicht Bytes. + +```ruby +write container(:fixed, file: 'out/cudo.txt', layout: LAYOUT, encoding: 'Windows-1252') +# KOPF20261009000001234.50 + +read container(:fixed, file: 'out/cudo.txt', layout: LAYOUT) +# { 'satyp' => 'KOPF', 'datum' => '20261009', 'summe' => '000001234.50' } +``` + +### 5.6 Layout-Typen und `validate`-Regeln + +Layout-Typen beschreiben das Textformat einer Datei mit festen Feldern; die Regel `type:` von `validate` (3.5) prüft +dagegen, ob sich ein Wert in einen Ruby-Typ umwandeln lässt. Beides lässt sich kombinieren: Rohdaten mit Regeln prüfen +und die fertigen Sätze gegen das Layout. + +| Layout-Typ | ähnliche Regel | Unterschied | +| --- | --- | --- | +| `AN` + Länge | `max: 10` | gleich | +| `N` | `type: :integer` bzw. `type: :decimal` | `:decimal` nimmt jedes Ruby-`Float` an (auch `1e3`, `.5`); `N` nur Ziffern mit optionalem Punkt, dafür mit Stellenzahl | +| `DAT` | `type: :date` | `:date` erwartet ISO 8601 (`2026-10-09`) oder ein `Date`, `DAT` dagegen `20261009` | +| `TIM` | `format: /\A\d{6}\z/` | `TIM` prüft zusätzlich Stunden, Minuten und Sekunden | + +### 5.7 Optionen + +| Option | Wirkung | | --- | --- | -| `read` | gelesene Quellzeilen | -| `written` | Datenzeilen nach allen Schritten (ohne Kopf- und Schlusssätze) | -| `output` | an die Ziele gegebene Zeilen inkl. Kopf- und Schlusssätzen | -| `rejected`, `errors` | verworfene Zeilen; die ersten 100 mit Schritt, Meldung und Zeile | -| `warnings` | Zeilen mit Warnung aus `validate …, warn: true` | -| `duration`, `started_at`, `finished_at` | Laufzeit | -| `error` | Exception eines abgebrochenen Laufs | -| `name` | Name aus `name`, über `etl` Default der Skriptname ohne `.rb` | -| `totals`, `total(spalte, :read \| :written)` | Summen aus `totals` als `Rational` (exakt) | -| `to_h` | alles als Hash, wie in der Laufhistorie | -| `ok?` | kein Abbruch und nichts verworfen | -| `to_s` | `read=9 written=4 rejected=4 0.097s`, mit `totals` zusätzlich `dmbtr=1500.25/1500.25` (Quelle/Ergebnis) | +| `type_field: 'satyp'` | Satzart aus dieser Spalte statt aus dem ersten Wert der Zeile | +| `strict: false` | Felder, die nicht im Layout stehen, beim Schreiben weglassen statt abbrechen; beim Lesen mehr Werte als Felder erlauben | +| `format: true` | Werte formatieren (5.3) | -Der Logger schreibt `start` und `finish` (Info), jede verworfene Zeile (Warnung) und einen Abbruch (Fehler). -`on_finish` läuft nach jedem Lauf, auch nach einem Abbruch; ein Fehler im Hook wird nur geloggt. +Eine Satzart, die nicht im Layout steht, ist immer ein Fehler (`record type "X" not in layout (DOHD, LARC)`). -```ruby -on_finish { |stats| DataPipeline::Slack.new(channel: '#etl').send_message("Buchungen: #{stats}") unless stats.ok? } +--- + +## 6. Betrieb + +### 6.1 Kommandozeile und Exit-Codes + +`etl` (im Repository `bin/etl`) startet ein Pipeline-Skript; `run` ist optional: + +```sh +bin/etl run export.rb --from=2025-11-01 --till=2025-11-30 +bin/etl run examples/kunden_import.rb --dry-run --limit=20 +bin/etl run pipeline.rb --param land=DE --out=data # im Skript: params[:land], params[:out] +bin/etl run cudo.rb --lock=data/.cudo.lock --history=data/runs.ndjson ``` -**Slack einrichten** – `DataPipeline::Slack` schickt Text über die Slack-API `chat.postMessage`. Voraussetzungen: +| Option | Wirkung | +| --- | --- | +| `--from`, `--till` | in `params[:from]`/`params[:till]` und unverändert an das Skript weitergereicht (für Skripte mit eigenem `OptionParser`) | +| `--param KEY=VALUE` | `params[:key]`, mehrfach möglich | +| `--dry-run`, `--limit=N` | für alle Pipelines des Skripts | +| `--lock=FILE` | Sperre für das ganze Skript (6.3) | +| `--history=FILE` | Laufhistorie für alle Pipelines des Skripts (6.6) | +| `--verbose`, `--quiet` | Log-Level Debug bzw. Warnung (Default: Info) | +| jede andere `--name=wert` | `params[:name]` und unverändert an das Skript, z. B. `--out=data`, `--incremental=1` | + +| Exit-Code | Bedeutung | +| --- | --- | +| `0` | alle Pipelines ohne Abbruch und ohne verworfene Zeilen | +| `1` | eine Pipeline ist abgebrochen (auch `ReconcileError`) oder das Skript hat einen Fehler | +| `2` | fertig, aber mit verworfenen Zeilen (`on_error :skip`/`:collect`) | +| `3` | ein anderer Lauf hält die Sperre; es wurde nichts gemacht | + +Ohne `etl` (mit `ruby skript.rb --out=x`) liest `params` die Argumente direkt aus `ARGV`. Das `Rakefile` der +Anwendung ruft `bin/etl` auf, z. B. `rake process:posting[2025-11-01,2025-11-30]`; ein fehlgeschlagener Lauf stoppt +Rake. + +### 6.2 Verbindungen + +Zugangsdaten stehen nicht in den Skripten, sondern einmal in `config/connections.yml` (relativ zum Arbeitsverzeichnis; +eine andere Datei über `DataPipeline.connections_file = …` oder `ENV['DATAPIPELINE_CONNECTIONS']`). Die Datei wird +mit ERB gelesen: Passwörter kommen aus der Umgebung, die Datei selbst kann ins Repository. `type` ist der +Plugin-Typ, alle anderen Schlüssel sind dessen Optionen. + +```yaml +# config/connections.yml +pds: + type: postgres + host: <%= ENV.fetch('PDS_HOST', 'localhost') %> + database: pds + username: etl + password: <%= ENV['PDS_PASSWORD'] %> +``` + +```ruby +read connection(:pds, query: 'select * from rechnung where datum >= $1', params: [params[:from]]) +write container(:postgres, connection: :pds, table: 'export_log') +``` + +Optionen beim Aufruf haben Vorrang vor der Datei. Eine unbekannte Verbindung, eine fehlende Datei oder ein Typ, der +nicht zu `container(:typ, …)` passt, bricht mit `ArgumentError` ab. Die Datei wird je Prozess einmal gelesen. Das +Gem lädt keine `.env`; Umgebungsvariablen setzt die Anwendung oder der Cronjob. + +### 6.3 Sperre gegen parallele Läufe + +Startet ein Cronjob, während der vorige Lauf noch arbeitet, lesen beide dasselbe Wasserzeichen und exportieren doppelt. +`lock` verhindert das mit einer exklusiven Sperre (`flock`) auf eine Datei. Ein zweiter Lauf wartet nicht, sondern +bricht sofort mit `DataPipeline::LockError` ab (Exit-Code `3`), schreibt nichts und ruft wie jeder Abbruch +`on_finish` auf. + +```ruby +lock 'data/.cudo.lock' # in der Pipeline: von run bis zum Ende des Laufs +``` + +```sh +bin/etl run cudo.rb --lock=data/.cudo.lock # für das ganze Skript, z. B. mehrere Pipelines nacheinander +``` + +Das Betriebssystem gibt die Sperre beim Ende des Prozesses frei, auch nach einem Absturz; es bleiben keine verwaisten +Sperren. Die Datei bleibt liegen und enthält die PID des letzten Laufs; die Meldung des zweiten Laufs nennt sie. Die +Sperre gilt auf einem Rechner (bei Netzlaufwerken nicht verlassen). Auch ein Dry-Run nimmt die Sperre und legt die +Datei an. + +### 6.4 Summen und Abgleich + +Bei einer Buchhaltungsschnittstelle reicht es nicht, dass der Lauf durchläuft: Die Summen müssen stimmen. `totals` +summiert Spalten über die Quellzeilen (`:read`) und über die Zeilen nach allen Schritten (`:written`, ohne Kopf- und +Schlusssätze), exakt als `Rational`. Leere und nicht numerische Werte zählen nicht mit; verworfene Zeilen zählen in +`:read`, aber nicht in `:written`. + +`reconcile` läuft nach allen Schritten und **vor dem Schreiben**. Gibt der Block eine Meldung (oder ein Array von +Meldungen) zurück, bricht der Lauf mit `DataPipeline::ReconcileError` ab und schreibt nichts (Exit-Code `1`, +`on_finish` läuft). `nil` heisst in Ordnung. Mehrere `reconcile`-Blöcke sind möglich, ihre Meldungen werden +zusammengefasst. + +```ruby +totals 'dmbtr' +reconcile do |stats| + quelle, ziel = stats.total('dmbtr', :read), stats.total('dmbtr') + "Summe dmbtr: Quelle #{DataPipeline.decimal(quelle)}, Ergebnis #{DataPipeline.decimal(ziel)}" if quelle != ziel +end +reconcile { |stats| 'keine Buchungen' if stats.written.zero? } +``` + +`DataPipeline.decimal(wert)` gibt eine Summe als exakten Dezimaltext aus (`1500.25` statt `6001/4`). Im Streaming +läuft `reconcile` nach der letzten Zeile, die Zeilen sind dann schon geschrieben; mit Postgres und +`batch_size: :all` werden sie beim Abbruch zurückgerollt (4.2). Das Beispiel 2.6 zeigt einen Saldo-Abgleich. + +### 6.5 Inkrementelle Läufe + +`watermark` merkt sich je Name den höchsten Wert einer Spalte in einer JSON-Datei. Die Pipeline verfolgt jede +Quellzeile und speichert den neuen Wert nur nach einem erfolgreichen Lauf (nicht nach einem Abbruch, nicht im +Dry-Run). Zahlen als Text werden numerisch verglichen. + +```ruby +wm = watermark :rechnungen, column: 'rechnung_id', file: 'data/.state.json', initial: 0 +read pg, query: 'select * from rechnung where rechnung_id > $1 order by rechnung_id', params: [wm.value] +``` + +`wm.value` ist der Stand vor dem Lauf, `wm.current` der höchste bisher gesehene Wert. Mit `limit` sollte die Quelle +nach der Spalte sortiert sein. Verworfene Zeilen schieben das Wasserzeichen mit; bei `:collect` liegen sie in der +Fehlerdatei. Gegen doppelte Läufe hilft die Sperre (6.3). + +### 6.6 Laufhistorie + +`history 'data/runs.ndjson'` (oder `etl … --history=data/runs.ndjson` für alle Pipelines eines Skripts) hängt nach +jedem Lauf eine JSON-Zeile an, auch nach einem Abbruch und im Dry-Run. Damit lässt sich später beantworten, wann der +letzte erfolgreiche Lauf war und wie viele Zeilen er hatte. Parallele Läufe schreiben sich nicht in die Quere; ein +Fehler beim Schreiben der Historie wird nur geloggt. + +```json +{"name":"cudo","version":"1.0","started_at":"2026-10-09T06:00:00.123+02:00","finished_at":"2026-10-09T06:00:04.336+02:00","duration":4.213,"read":1250,"written":1243,"output":1245,"rejected":7,"warnings":0,"dry_run":false,"ok":false,"error":null,"totals":{"dmbtr":{"read":"98231.4","written":"97810.1"}}} +``` + +```ruby +letzter = DataPipeline.history('data/runs.ndjson', name: 'cudo').reverse.find { |run| run['ok'] && !run['dry_run'] } +``` + +### 6.7 Dry-Run und Limit + +Im Dry-Run läuft alles, aber nichts wird geschrieben: kein Ziel, keine Fehlerdatei, kein Wasserzeichen; Dateien und +SQLite-Datenbanken werden nicht angelegt. Am Ende stehen die Zusammenfassung und die ersten Zeilen (`preview`, Default +10) als JSON auf `$stdout`. Geschrieben werden nur die Laufhistorie (mit `"dry_run": true`) und die Sperrdatei. +Eigene Schritte mit Nebenwirkungen (API-Aufrufe in einem `map`) fragen `dry_run?` ab. + +`limit n` verarbeitet nur die ersten n Quellzeilen; im Streaming wird das Lesen dann beendet. + +### 6.8 Benachrichtigung über Slack + +`DataPipeline::Slack` schickt Text über die Slack-API `chat.postMessage`. Einrichten: 1. In Slack eine App anlegen (api.slack.com/apps), unter *OAuth & Permissions* den Bot-Scope `chat:write` vergeben, die App im Workspace installieren und den *Bot User OAuth Token* (`xoxb-…`) kopieren. 2. Die App in den Ziel-Channel einladen (`/invite @appname`), sonst antwortet Slack mit `not_in_channel`. -3. Das Gem `http` in das Gemfile der Anwendung aufnehmen (siehe 2.6); die Klasse wird erst beim ersten Zugriff geladen. -4. Den Token als Umgebungsvariable `SLACK_TOKEN` setzen, z. B. in der `.env` der Anwendung (das Gem lädt keine - `.env`), oder direkt mit `token:` übergeben. Den Token nicht ins Pipeline-Skript schreiben. +3. Das Gem `http` ins Gemfile der Anwendung aufnehmen (4.10). +4. Den Token als Umgebungsvariable `SLACK_TOKEN` setzen oder mit `token:` übergeben; nicht ins Skript schreiben. ```ruby slack = DataPipeline::Slack.new # Token aus ENV['SLACK_TOKEN'], Channel '#general' @@ -904,9 +1445,9 @@ slack.send_message('Test aus der Pipeline', verbose: true) # verbose `on_finish` wird sie nur geloggt (`on_finish: RuntimeError: Slack#send_message failed: …`), der Lauf und sein Exit-Code bleiben unverändert. Es gibt keine Wiederholung bei Netzwerkfehlern. -Ein ausführlicheres Beispiel mit Status, Zählern, Laufzeit und den ersten verworfenen Zeilen. Der Notifier wird vor -der Pipeline angelegt; ein fehlender Token fällt so schon beim Start auf und nicht erst nach dem Lauf. Slack setzt -`*fett*`, `_kursiv_` und `` `code` `` um. +Eine ausführliche Meldung mit Status, Zählern, Laufzeit und den ersten verworfenen Zeilen. Der Notifier wird vor der +Pipeline angelegt, damit ein fehlender Token schon beim Start auffällt. Slack setzt `*fett*`, `_kursiv_` und +`` `code` `` um. ```ruby slack = DataPipeline::Slack.new(channel: '#etl') @@ -950,253 +1491,139 @@ gelesen 1250 · geschrieben 1243 · verworfen 7 · Warnungen 0 … und 2 weitere, siehe data/fehler.csv ``` -`stats.errors` enthält höchstens 100 Einträge als Hash mit `step`, `error` und der Eingangszeile `row`. `params` gibt -es im Pipeline-Block; ausserhalb heisst es `DataPipeline.params`. +`params` gibt es im Pipeline-Block; ausserhalb heisst es `DataPipeline.params`. -**Inkrementelle Läufe** – `watermark` merkt sich je Name den höchsten Wert einer Spalte in einer JSON-Datei. Die -Pipeline verfolgt jede Quellzeile und speichert den neuen Wert nur nach einem erfolgreichen Lauf (nicht nach Abbruch, -nicht im Dry-Run). Zahlen als Text werden numerisch verglichen. Mit `limit` sollte die Quelle nach der Spalte -sortiert sein; verworfene Zeilen schieben das Wasserzeichen mit (sie liegen bei `:collect` in der Fehlerdatei). +### 6.9 Fehlersuche -```ruby -wm = watermark :rechnungen, column: 'rechnung_id', file: 'data/.state.json', initial: 0 -read pg, query: 'select * from rechnung where rechnung_id > $1 order by rechnung_id', params: [wm.value] -``` - -**Verbindungen** – Zugangsdaten stehen nicht in den Skripten, sondern einmal in `config/connections.yml` (oder der -Datei aus `DataPipeline.connections_file = …` bzw. `ENV['DATAPIPELINE_CONNECTIONS']`). Die Datei wird mit ERB -gelesen, Passwörter kommen so aus der Umgebung und die Datei kann ins Repository. `type` ist der Plugin-Typ, alle -anderen Schlüssel sind dessen Optionen; Optionen beim Aufruf haben Vorrang. - -```yaml -# config/connections.yml -pds: - type: postgres - host: <%= ENV.fetch('PDS_HOST', 'localhost') %> - database: pds - username: etl - password: <%= ENV['PDS_PASSWORD'] %> -``` - -```ruby -read connection(:pds, query: 'select * from rechnung where datum >= $1', params: [params[:from]]) -write container(:postgres, connection: :pds, table: 'export_log') # Typ muss zur Verbindung passen -``` - -Eine unbekannte Verbindung, eine fehlende Datei oder ein abweichender Typ bricht mit `ArgumentError` ab. Die Datei -wird je Prozess einmal gelesen. - -**Summen und Abgleich** – Bei einer Buchhaltungsschnittstelle reicht es nicht, dass der Lauf durchläuft: Die Summen -müssen stimmen. `totals` summiert Spalten über die Quellzeilen (`:read`) und über die Zeilen nach allen Schritten -(`:written`, ohne Kopf- und Schlusssätze), exakt als `Rational`. Leere und nicht numerische Werte zählen nicht mit. -`reconcile` läuft nach allen Schritten und **vor dem Schreiben**; gibt der Block eine Meldung zurück, bricht der Lauf -mit `DataPipeline::ReconcileError` ab und schreibt nichts (Exit-Code `1`, `on_finish` läuft). Im Streaming läuft -`reconcile` nach der letzten Zeile, die Zeilen sind dann schon geschrieben; mit Postgres und `batch_size: :all` werden -sie beim Abbruch zurückgerollt (2.5). - -```ruby -totals 'dmbtr' -reconcile do |stats| - quelle, ziel = stats.total('dmbtr', :read), stats.total('dmbtr') - "Summe dmbtr: Quelle #{DataPipeline.decimal(quelle)}, Ergebnis #{DataPipeline.decimal(ziel)}" if quelle != ziel -end -reconcile { |stats| 'keine Buchungen' if stats.written.zero? } -``` - -`DataPipeline.decimal(wert)` gibt eine Summe als exakten Dezimaltext aus (`1500.25` statt `6001/4`). - -**Laufhistorie** – `history 'data/runs.ndjson'` (oder `etl … --history=data/runs.ndjson` für alle Pipelines eines -Skripts) hängt nach jedem Lauf eine JSON-Zeile an, auch nach einem Abbruch und im Dry-Run. Damit lässt sich später -beantworten, wann der letzte erfolgreiche Lauf war und wie viele Zeilen er hatte. Ein Fehler beim Schreiben der -Historie wird nur geloggt. - -```json -{"name":"cudo","version":"1.0","started_at":"2026-10-09T06:00:00.123+02:00","finished_at":"2026-10-09T06:00:04.336+02:00","duration":4.213,"read":1250,"written":1243,"output":1245,"rejected":7,"warnings":0,"dry_run":false,"ok":false,"error":null,"totals":{"dmbtr":{"read":"98231.4","written":"97810.1"}}} -``` - -```ruby -letzter = DataPipeline.history('data/runs.ndjson', name: 'cudo').reverse.find { |run| run['ok'] } -``` - -**Sperre gegen parallele Läufe** – Startet ein Cronjob, während der vorige Lauf noch arbeitet, lesen beide dasselbe -Wasserzeichen und exportieren doppelt. `lock` verhindert das mit einer exklusiven Sperre (`flock`) auf eine Datei: -Ein zweiter Lauf wartet nicht, sondern bricht sofort mit `DataPipeline::LockError` ab (Exit-Code `3`), schreibt nichts -und ruft wie jeder Abbruch `on_finish` auf. Das Betriebssystem gibt die Sperre beim Ende des Prozesses frei, auch nach -einem Absturz; es bleiben also keine verwaisten Sperren. Die Datei bleibt liegen und enthält die PID des letzten Laufs. - -```ruby -lock 'data/.cudo.lock' # in der Pipeline: von run bis zum Ende des Laufs -``` - -```sh -bin/etl run cudo.rb --lock=data/.cudo.lock # für das ganze Skript, z. B. mehrere Pipelines nacheinander -``` - -**Dry-Run und Limit** – Im Dry-Run läuft alles, aber nichts wird geschrieben: kein Ziel, keine Fehlerdatei, kein -Wasserzeichen; eine CSV-Datei wird nicht angelegt. Am Ende stehen die Zusammenfassung und die ersten Zeilen -(`preview`, Default 10) als JSON auf `$stdout`. `limit n` verarbeitet nur die ersten n Quellzeilen, im Streaming -wird das Lesen dann beendet. - -**Kommandozeile** – `bin/etl` startet ein Pipeline-Skript; `run` ist optional (`bin/etl skript.rb` genügt): - -```sh -bin/etl run export.rb --from=2025-11-01 --till=2025-11-30 -bin/etl run examples/kunden_import.rb --dry-run --limit=20 -bin/etl run pipeline.rb --param land=DE # im Skript: params[:land] -``` - -| Option | Wirkung | -| --- | --- | -| `--from`, `--till` | in `params[:from]`/`params[:till]` und unverändert an das Skript weitergereicht (eigener `OptionParser` der Importer) | -| `--param KEY=VALUE` | `params[:key]`, mehrfach möglich | -| `--dry-run`, `--limit=N` | für alle Pipelines des Skripts | -| `--lock=FILE` | Sperre für das ganze Skript; ist sie belegt, endet `etl` sofort mit Exit-Code `3` | -| `--history=FILE` | Laufhistorie für alle Pipelines des Skripts; als `name` gilt der Skriptname | -| `--verbose`, `--quiet` | Log-Level Debug bzw. Warnung (Default über `bin/etl`: Info) | -| jede andere `--name=wert` | `params[:name]` und unverändert an das Skript, z. B. `--out=data`, `--incremental=1` | - -Exit-Code: `0` ok, `1` Abbruch, `2` fertig, aber mit verworfenen Zeilen, `3` ein anderer Lauf hält die Sperre. Das `Rakefile` ruft `bin/etl` auf: -`rake process:posting[2025-11-01,2025-11-30]`, `rake process:all[2025-11-01,2025-11-30,dry]`. Ein fehlgeschlagener Lauf stoppt Rake. - -### 2.8 Tests - -`rake test` (oder `bundle exec rake`) startet die Minitest-Suite in [test/](test/): rund 205 Fälle für Plugins, Kern, -Streaming, Betrieb, Kommandozeile und die Beispiele. - -| Ordner | Inhalt | -| --- | --- | -| `test/plugins` | je Plugin das gemeinsame Testkit `test/support/storage_contract.rb` (Schreiben und Zurücklesen, zeilenweise, als Quelle und Ziel einer Pipeline) plus eigene Fälle; REST gegen einen lokalen Testserver; Laden der Gems erst bei Nutzung in einem frischen Prozess | -| `test/core` | DSL, Einstellungen, Plugin-Registry, Parallel- und Verkettungsmodus, Streaming, Lookups, mehrere Ziele, Betrieb; `compat_test.rb` hält die Ausgabe des `transform`-Modells von v1.2.1 fest | -| `test/cli` | `bin/etl` als Prozess: Weitergabe von `--from`/`--till`, Parameter, Dry-Run, Exit-Codes | -| `test/examples` | die Beispiele aus Abschnitt 1 (Code in der Doku = Datei, Lauf und Ergebnis; REST gegen die Demo-API) | - -Die `.env` wird in Tests nicht geladen; kein Test erreicht die Produktivdatenbank. Postgres-Tests legen je Test eine -Wegwerf-Datenbank auf einem lokalen Server an (`ETL_TEST_PG_HOST`, `_PORT`, `_USER`, `_PASSWORD`; Default -`localhost`, `$USER`). Ohne erreichbaren Server werden sie übersprungen (23 Tests). Ein eigenes Plugin lässt sich mit -`MemoryStorage` aus `test/test_helper.rb` als Quelle oder Ziel testen; mit `include StorageContract` und einer -Methode `contract_storage(env, :write | :read)` bekommt es die gemeinsamen Tests. - -### 2.9 Satzarten-Layouts - -Dateien mit mehreren Satzarten (Kopf, Positionen, Schluss) haben je Satzart eine feste Feldliste. `DataPipeline::Layout` -beschreibt sie; die Satzart einer Zeile ist ihr erster Wert (oder `type_field:`), Schlüssel werden ohne Rücksicht auf -Gross-/Kleinschreibung zugeordnet. - -```ruby -LAYOUT = DataPipeline::Layout.new do - record 'DOHD', [['SATYP', 'AN', 4, true], ['LFDNR', 'N', 10, true], ['BKTXT', 'AN', 25, true], ...] - record 'LARC', [['SATYP', 'AN', 4, true], ['SUMME', 'N', 10, true]] -end - -validate LAYOUT, warn: true # Pflicht, Länge, Format je Feld -write container(:csv, file: 'x.ok', layout: LAYOUT, csv: { headers: false, col_sep: '|' }) -``` - -**Felddefinition** – jedes Feld ist ein Array `[Name, Typ, Länge, Pflicht]`: - -| Teil | Bedeutung | -| --- | --- | -| Name | Spaltenname; in der Zeile ohne Rücksicht auf Gross-/Kleinschreibung, als String oder Symbol gesucht | -| Typ | `AN`, `N`, `DAT` oder `TIM` (siehe unten), Gross-/Kleinschreibung egal; fehlt er (`nil`), gilt `AN` | -| Länge | Zahl oder bei Nachkommastellen ein String (`'11.2'`); fehlt sie, wird die Länge nicht geprüft | -| Pflicht | `true` = darf nicht leer sein; fehlt sie, ist das Feld optional | - -**Datentypen** – geprüft wird immer der Wert als Text (`to_s`). Ein leerer Wert (`nil`, `''`, nur Leerzeichen) -wird nur auf Pflicht geprüft, nie auf den Typ. - -| Typ | Inhalt | Gültig | Länge | Meldung | -| --- | --- | --- | --- | --- | -| `AN` | Text (alphanumerisch) | jeder Wert | höchstens `Länge` Zeichen | `NAME: longer than 10` | -| `N` | Zahl | `\A[+-]?\d+(\.\d+)?\z`: Ziffern, optional Vorzeichen und Punkt als Dezimaltrenner | `10` = höchstens 10 Stellen, keine Nachkommastellen; `'11.2'` = höchstens 9 Vor- und 2 Nachkommastellen (Gesamtlänge 11 wie in SAP); das Vorzeichen zählt nicht, führende Nullen schon | `NAME: not a number`, `… more than 9 digits`, `… more than 2 decimals` | -| `DAT` | Datum | genau 8 Ziffern `YYYYMMDD`, gültiges Kalenderdatum | wird nicht geprüft | `NAME: not a date YYYYMMDD` | -| `TIM` | Uhrzeit | genau 6 Ziffern `HHMMSS`, 00:00:00 bis 23:59:59 | wird nicht geprüft | `NAME: not a time HHMMSS` | - -Ein unbekannter Typ wird wie `AN` behandelt. - -**Formatieren** – Ohne Option prüft das Layout nur: Werte werden unverändert geschrieben, ein `Date` würde also als -`2026-10-09` geprüft und abgelehnt. Mit `Layout.new(format: true)` bringt das Layout die Werte vor dem Prüfen und -Schreiben selbst in das Textformat ihres Typs. Geprüft wird dann genau der Text, der in die Datei kommt. - -| Typ | Eingabe | Ergebnis | +| Meldung | Ursache | Abhilfe | | --- | --- | --- | -| `N` | Integer, Float, Rational, BigDecimal | Nachkommastellen aus der Länge (`11.2` → 2), kaufmännisch gerundet: `12.345` → `12.35`, `-0.005` → `-0.01`, `7` → `7.00`; ein Float zählt mit seinem kürzesten Text, `12.345` also nicht als `12.3449999…` | -| `N` | Zahl als Text | auf die Nachkommastellen aufgefüllt (`'3.5'` → `3.50`); zu viele Nachkommastellen bleiben stehen und werden gemeldet, Text wird nie gerundet | -| `DAT` | `Date`, `Time`, `'2026-10-09'` | `20261009` | -| `TIM` | `Time`, `'07:05:03'` | `070503` | -| alle | `nil` | bleibt leer | +| `Unsupported storage type or class missing: XyzStorage` | Tippfehler in `container(:xyz)` oder eigenes Plugin nicht registriert | Typ prüfen, `DataPipeline.register` (4.1) | +| `… needs the gem 'pg': add gem 'pg' to the Gemfile` | Plugin-Gem fehlt | Gem ins Gemfile der Anwendung, `bundle install` (4.10) | +| `rejected by validate#2: …` | Zeile hat eine Regel verletzt | Fehlerdatei (`on_error :collect`) ansehen; Schritt-Nummer = Reihenfolge im Skript | +| `DataPipeline::LockError: … is locked by pid 4711` | ein anderer Lauf läuft noch (Exit-Code `3`) | warten; hängt der Prozess, ihn beenden – die Sperre löst sich dann von selbst | +| `DataPipeline::ReconcileError: …` | ein `reconcile`-Block hat eine Abweichung gemeldet | Quelle und Schritte prüfen; es wurde nichts geschrieben (ausser im Streaming) | +| `DataPipeline::LayoutError: X: fields not in layout: foo` | Zeile hat ein Feld, das die Satzart nicht kennt | Spalte umbenennen (`rename`) oder weglassen (`select`), Layout ergänzen oder `strict: false` | +| `DataPipeline::LayoutError: record type "X" not in layout` | erster Wert (bzw. `type_field`) ist keine bekannte Satzart | Reihenfolge der Spalten prüfen; `type_field:` setzen | +| `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 | -Anderes bleibt unverändert und wird wie gewohnt geprüft. Führende Nullen ergänzt erst die Festbreiten-Datei (unten); -in CSV braucht es dafür weiterhin ein `map` (`format('%06d', …)`). - -```ruby -LAYOUT = DataPipeline::Layout.new(format: true) do - record 'KOPF', [['SATYP', 'AN', 4, true], ['DATUM', 'DAT', 8, true], ['SUMME', 'N', '11.2', true]] -end -LAYOUT.values({ 'satyp' => 'KOPF', 'datum' => Date.new(2026, 10, 9), 'summe' => 1234.5 }) -# => ["KOPF", "20261009", "1234.50"] -``` - -**Festbreiten-Dateien** – `container(:fixed, file:, layout:)` (`DataPipeline::FixedStorage`) schreibt und liest -Dateien ohne Trennzeichen, in denen jedes Feld eine feste Anzahl Zeichen hat. Jedes `AN`- und `N`-Feld braucht dafür -eine Länge, sonst bricht schon `container` mit `LayoutError` ab. - -| Typ | Breite | Ausrichtung | -| --- | --- | --- | -| `AN` | Länge | links, rechts mit Leerzeichen aufgefüllt | -| `N` | Vor- + Nachkommastellen + Punkt: `'11.2'` → 12, `5` → 5 | rechts, links mit Nullen aufgefüllt, Vorzeichen vorne: `-0000012.50` | -| `DAT`, `TIM` | 8 bzw. 6 | genau so viele Ziffern | -| leeres Feld | Breite des Felds | nur Leerzeichen | - -Beim Schreiben werden die Werte immer formatiert (wie `format: true`). Ein Wert, der länger als sein Feld ist, wird -nicht abgeschnitten, sondern bricht mit `LayoutError` ab; mit `validate LAYOUT` vorher fällt er schon als Zeilenfehler -auf. Beim Lesen erkennt das Plugin die Satzart an ihrem Feld, die Schlüssel sind die Feldnamen klein geschrieben, die -Werte ohne Auffüllung (Zahlen behalten ihre führenden Nullen), leere Felder sind `nil`. Breiten zählen Zeichen, nicht -Bytes. Weitere Optionen: `encoding`, `replace` (wie CSV, 2.5) und `line_end: "\r\n"`. - -```ruby -write container(:fixed, file: 'out/cudo.txt', layout: LAYOUT, encoding: 'Windows-1252') -# KOPF20261009000001234.50 - -read container(:fixed, file: 'in/rueckmeldung.txt', layout: LAYOUT) -# { 'satyp' => 'KOPF', 'datum' => '20261009', 'summe' => '000001234.50' } -``` - -**Lesen mit Layout** – Auch eine CSV-Datei mit Satzarten lässt sich mit `layout:` wieder einlesen (nur mit -`headers: false`). Jede Zeile bekommt die Feldnamen ihrer Satzart als Schlüssel; eine unbekannte Satzart oder mehr -Werte als Felder bricht mit `LayoutError` ab. - -```ruby -read container(:csv, file: 'in/cudo.ok', layout: LAYOUT, csv: { headers: false, col_sep: '|' }) -``` - -**Verhältnis zu `validate`-Regeln** – Layout-Typen beschreiben das Textformat einer Datei mit festen Feldern, die -Regel `type:` von `validate` (2.7) prüft dagegen, ob sich ein Wert in einen Ruby-Typ umwandeln lässt. Beides lässt -sich kombinieren, z. B. Rohdaten mit Regeln prüfen und die fertigen Sätze gegen das Layout. - -| Layout-Typ | ähnliche `validate`-Regel | Unterschied | -| --- | --- | --- | -| `AN` + Länge | `max: 10` | gleich | -| `N` | `type: :integer` bzw. `type: :decimal` | `:decimal` nimmt jedes Ruby-`Float` an (auch `1e3`, `.5`); `N` nur Ziffern mit optionalem Punkt, dafür mit Stellenzahl | -| `DAT` | `type: :date` | `:date` erwartet ISO 8601 (`2026-10-09`) oder ein `Date`, `DAT` dagegen `20261009` | -| `TIM` | `format: /\A\d{6}\z/` | `TIM` prüft zusätzlich Stunden, Minuten und Sekunden | - -**Optionen** – `Layout.new(type_field: 'satyp')` nimmt die Satzart aus dieser Spalte statt aus dem ersten Wert der -Zeile; eine Satzart, die nicht im Layout steht, ist ein Fehler (`record type "X" not in layout (…)`). -`strict: false` lässt beim Schreiben Felder zu, die nicht im Layout stehen (sie werden weggelassen). -`format: true` formatiert die Werte (siehe oben). - -Beim Schreiben stehen die Werte in der Reihenfolge des Layouts, fehlende Felder bleiben leer, ein unbekanntes Feld -bricht mit `DataPipeline::LayoutError` ab. Eine falsch benannte SQL-Spalte verschiebt also nicht mehr still alle -folgenden Felder. Ein vollständiges Beispiel sind die CUDO-Satzarten der SAP-Schnittstelle im Projekt `etl` (`cudo/layout.rb`), -erzeugt aus der Schnittstellenbeschreibung. +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. --- -## 3. Features und Verbesserungen +## 7. Veraltetes und Umstieg -Ziel: aus dem Exportwerkzeug ein generisches, testbares ETL-Framework machen. Reihenfolge = empfohlene Umsetzung. +### 7.1 Von 2.0 auf 2.1 -### 3.1 Erweiterbarkeit +2.1 ist abwärtskompatibel; bestehende Skripte laufen ohne Änderung. Anders als bisher: -1. **Weitere Plugins:** SFTP-Ablage. -2. **Veröffentlichung:** eigenes Remote-Repository für das Gem, Lizenz festlegen, Versionen taggen. +| Thema | 2.0 | 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 + +### 8.1 Aufbau + +| Datei | Inhalt | +| --- | --- | +| [lib/datapipeline.rb](lib/datapipeline.rb) | DSL und Ausführung (`run`), `Stats`, `Context`, Kurzschritte, Summen, Historie | +| [lib/datapipeline/validation.rb](lib/datapipeline/validation.rb) | Regeln für `validate` | +| [lib/datapipeline/layout.rb](lib/datapipeline/layout.rb) | Satzarten-Layouts: prüfen, formatieren, feste Breite, lesen | +| [lib/datapipeline/watermark.rb](lib/datapipeline/watermark.rb) | Wasserzeichen für inkrementelle Läufe | +| [lib/datapipeline/lock.rb](lib/datapipeline/lock.rb) | Sperre gegen parallele Läufe | +| [lib/datapipeline/connections.rb](lib/datapipeline/connections.rb) | Verbindungsdatei | +| [lib/datapipeline/storage/](lib/datapipeline/storage/) | `storage_plugin.rb` (Basisklasse) und je Plugin eine Datei | +| [lib/datapipeline/notification/slack.rb](lib/datapipeline/notification/slack.rb) | Slack | +| [lib/datapipeline/version.rb](lib/datapipeline/version.rb) | `DataPipeline::VERSION` | +| [exe/etl](exe/etl) | Kommando `etl`; [bin/etl](bin/etl) startet es im Repository mit dem Bundle | +| [examples/](examples/) | Beispiele aus Kapitel 2, Daten in `examples/data/`, Verbindungsdatei in `examples/config/`, Demo-API | +| [datapipeline.gemspec](datapipeline.gemspec), [CHANGELOG.md](CHANGELOG.md) | Gem-Beschreibung, Versionsverlauf | + +### 8.2 Tests + +`bundle exec rake test` startet die Minitest-Suite in [test/](test/), rund 205 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/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 | + +Eine `.env` wird in Tests nicht geladen; kein Test erreicht eine Produktivdatenbank. Postgres-Tests legen je Test eine +Wegwerf-Datenbank auf einem lokalen Server an (`ETL_TEST_PG_HOST`, `_PORT`, `_USER`, `_PASSWORD`; Default +`localhost`, Port 5432, `$USER`). Ohne erreichbaren Server werden sie übersprungen (23 Tests). Mit Postgres.app geht +ein eigener Testserver so, ohne die Daten der App anzufassen: + +```sh +initdb -D tmp/pg -A trust && pg_ctl -D tmp/pg -o "-p 54329 -k ''" -l tmp/pg.log start +ETL_TEST_PG_PORT=54329 bundle exec rake test +pg_ctl -D tmp/pg stop +``` + +Ein eigenes Plugin lässt sich mit `MemoryStorage` aus `test/test_helper.rb` als Quelle oder Ziel testen; mit +`include StorageContract` und einer Methode `contract_storage(env, :write | :read)` bekommt es die gemeinsamen Tests. + +### 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`). +- 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`. +- Beispiele in Kapitel 2 werden mit einem Marker `` vor dem Codeblock eingebunden; + der Doku-Test vergleicht Block und Datei, Änderungen also immer in beiden machen. + +--- + +## 9. Ausblick + +Offene Punkte, Reihenfolge = empfohlene Umsetzung: + +1. **Kommandozeile:** `etl check skript.rb` (Plugins, Gems und Verbindungen prüfen, ohne zu lesen oder zu + schreiben), `etl list` (Pipelines, Wasserzeichen, letzter Lauf aus der Historie), `--json` für die Stats. +2. **Qualität:** CI auf git.wicom.li mit `rake test` und Postgres-Service, RuboCop, `DataPipeline::Testing` mit + `MemoryStorage` als öffentliche Testhilfe für Anwendungen. +3. **SQLite im Streaming:** `write_row` wie bei Postgres in Paketen schreiben. +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. diff --git a/examples/buchungen_export.rb b/examples/buchungen_export.rb new file mode 100644 index 0000000..066e58a --- /dev/null +++ b/examples/buchungen_export.rb @@ -0,0 +1,59 @@ +# +# Produktiver Lauf: Buchungen prüfen, Summe und Saldo abgleichen, wiederholbar in die Buchhaltungs-Datenbank +# und als CSV in Windows-1252 für das Zielsystem schreiben; mit Sperre, Laufhistorie und Slack bei Problemen +# +# bin/etl run examples/buchungen_export.rb [--out=tmp] [--dry-run] +# +require 'fileutils' +require 'datapipeline' + +DATA = File.expand_path('data', __dir__) +out = DataPipeline.params.fetch(:out, 'tmp') +FileUtils.mkdir_p(out) + +# Zugangsdaten stehen nicht im Skript, sondern in der Verbindungsdatei (Default: config/connections.yml) +DataPipeline.connections_file = File.expand_path('config/connections.yml', __dir__) + +# Slack nur, wenn ein Token gesetzt ist; so fällt ein fehlender Token schon vor dem Lauf auf +slack = DataPipeline::Slack.new(channel: '#etl') if ENV['SLACK_TOKEN'] + +DataPipeline.new do + name 'buchungen_export' + lock "#{out}/.buchungen.lock" # läuft schon ein Export, bricht dieser ab (Exit-Code 3) + history "#{out}/runs.ndjson" # eine JSON-Zeile je Lauf, auch nach Abbruch + + read container(:csv, file: "#{DATA}/buchungen.csv", csv: { col_sep: ';' }) + + # 1. prüfen: ohne on_error bricht ein Fehler den Lauf ab, es entsteht keine halbe Buchung + validate 'belnr' => { required: true, format: /\A\d{10}\z/ }, + 'konto' => { required: true, format: /\A\d+\z/ }, + 'betrag' => { required: true, type: :decimal } + + # 2. umbauen mit Kurzschritten und einem map + default 'text' => 'ohne Text' + rename 'text' => 'buchungstext' + cast 'buzei' => :integer, 'datum' => :date, 'betrag' => :decimal + map do |row| + row.merge('shkzg' => row['betrag'].negative? ? 'H' : 'S', 'dmbtr' => format('%.2f', row['betrag'].abs)) + end + select %w[belnr buzei datum konto shkzg dmbtr betrag buchungstext] + + # 3. abgleichen, bevor etwas geschrieben wird + totals 'betrag' + reconcile do |stats| + quelle, ergebnis = stats.total('betrag', :read), stats.total('betrag') + next "Summe geändert: Quelle #{DataPipeline.decimal(quelle)}, Ergebnis #{DataPipeline.decimal(ergebnis)}" if quelle != ergebnis + + "Soll und Haben ungleich, Saldo #{DataPipeline.decimal(ergebnis)}" unless ergebnis.zero? + end + + # 4. schreiben: Upsert macht den Lauf wiederholbar; die CSV-Datei im Zeichensatz des Zielsystems + write connection(:buchhaltung) + write container(:csv, file: "#{out}/buchungen_sap.csv", encoding: 'Windows-1252', csv: { col_sep: ';' }, + columns: %w[belnr buzei datum konto shkzg dmbtr buchungstext]) + + on_finish do |stats| + puts "#{stats.name}: #{stats}" + slack&.send_message("*#{stats.name}*: #{stats}") unless stats.ok? + end +end.run diff --git a/examples/config/connections.yml b/examples/config/connections.yml new file mode 100644 index 0000000..e7c4797 --- /dev/null +++ b/examples/config/connections.yml @@ -0,0 +1,8 @@ +# Verbindungen der Beispiele. Die Datei wird mit ERB gelesen: Zugangsdaten und Pfade kommen aus der Umgebung, +# bei Postgres z. B. password: <%= ENV['PDS_PASSWORD'] %> +buchhaltung: + type: sqlite + database: <%= ENV.fetch('BUCHHALTUNG_DB', 'tmp/buchhaltung.sqlite3') %> + table: buchungen + create: true + upsert: [belnr, buzei] diff --git a/examples/data/buchungen.csv b/examples/data/buchungen.csv new file mode 100644 index 0000000..94df7cf --- /dev/null +++ b/examples/data/buchungen.csv @@ -0,0 +1,8 @@ +belnr;buzei;datum;konto;betrag;text +5100000001;1;2026-10-01;1020;1250.00;Zahlung Müller GmbH +5100000001;2;2026-10-01;1100;-1250.00;Zahlung Müller GmbH +5100000002;1;2026-10-02;6500;380.50;Büromaterial +5100000002;2;2026-10-02;1170;29.30;Vorsteuer +5100000002;3;2026-10-02;2000;-409.80;Papeterie Zürcher AG +5100000003;1;2026-10-03;3200;99.95; +5100000003;2;2026-10-03;1020;-99.95;Gutschrift diff --git a/test/examples/doc_examples_test.rb b/test/examples/doc_examples_test.rb index 42ee237..cb5f4a1 100644 --- a/test/examples/doc_examples_test.rb +++ b/test/examples/doc_examples_test.rb @@ -4,15 +4,15 @@ require 'open3' require 'fileutils' # -# the examples of README.md section 1: the code in the documentation is the file, +# the examples of README.md section 2: the code in the documentation is the file, # and every example runs and produces the documented result # class DocExamplesTest < PipelineTest DOC = File.read(File.join(ROOT, 'README.md')) EXAMPLES = DOC.scan(/\n```ruby\n(.*?)\n```/m) - def run_example dir, name, *args - env = { 'BUNDLE_GEMFILE' => File.join(ROOT, 'Gemfile') } + def run_example dir, name, *args, env: {} + env = { 'BUNDLE_GEMFILE' => File.join(ROOT, 'Gemfile') }.merge(env) Open3.capture3(env, RbConfig.ruby, File.join(ROOT, 'bin/etl'), 'run', File.join(ROOT, name), "--out=#{dir}", *args, chdir: dir) end @@ -20,7 +20,7 @@ class DocExamplesTest < PipelineTest def test_documented_code_equals_files assert_equal %w[examples/kunden_import.rb examples/rechnungen_satzarten.rb examples/bewegungen_inkrementell.rb examples/eigenes_plugin.rb examples/rest_lesen.rb examples/rest_schreiben.rb - examples/rest_zu_rest.rb], EXAMPLES.map(&:first) + examples/rest_zu_rest.rb examples/buchungen_export.rb], EXAMPLES.map(&:first) EXAMPLES.each do |file, code| assert_equal File.read(File.join(ROOT, file)).rstrip, code.rstrip, "#{file} differs from README.md" end @@ -152,4 +152,33 @@ class DocExamplesTest < PipelineTest end end end + + def test_buchungen_export + with_tmpdir do |dir| + env = { 'BUCHHALTUNG_DB' => "#{dir}/buchhaltung.sqlite3", 'SLACK_TOKEN' => nil } + out, err, status = run_example(dir, 'examples/buchungen_export.rb', '--dry-run', env:) + assert_equal 0, status.exitstatus, err + refute File.exist?("#{dir}/buchhaltung.sqlite3") + + 2.times do + out, err, status = run_example(dir, 'examples/buchungen_export.rb', env:) + assert_equal 0, status.exitstatus, err + end + assert_includes out, 'buchungen_export: read=7 written=7 rejected=0 betrag=0/0' + + db = DataPipeline::SqliteStorage.new(database: "#{dir}/buchhaltung.sqlite3", query: 'select * from buchungen order by belnr, buzei') + rows = db.read + db.close + assert_equal 7, rows.size # upsert: the second run updates instead of inserting + assert_equal({ 'belnr' => '5100000002', 'buzei' => 3, 'datum' => '2026-10-02', 'konto' => '2000', 'shkzg' => 'H', + 'dmbtr' => '409.80', 'betrag' => -409.8, 'buchungstext' => 'Papeterie Zürcher AG' }, rows[4]) + assert_equal 'ohne Text', rows[5]['buchungstext'] + + csv = File.read("#{dir}/buchungen_sap.csv", encoding: 'Windows-1252').encode('UTF-8') + assert_includes csv, "5100000001;1;2026-10-01;1020;S;1250.00;Zahlung Müller GmbH\n" + history = DataPipeline.history("#{dir}/runs.ndjson", name: 'buchungen_export') + assert_equal [true, true, true], history.map { |run| run['ok'] } + assert_equal [true, false, false], history.map { |run| run['dry_run'] } + end + end end