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