datapipeline/README.md

# DataPipeline – ETL-Framework in Ruby

Stand: 2026-10-09 · Gem `datapipeline` 3.2.0 ([CHANGELOG.md](CHANGELOG.md)) · Ruby 3.4 · Lizenz BSD-3-Clause ([LICENSE](LICENSE))

`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, Benachrichtigungen, Fehlersuche
7. [Umstieg](#7-umstieg) – auf 3.1, von 2.x auf 3.0 (`transform` entfernt), von 2.0 auf 2.1
8. [Entwicklung](#8-entwicklung) – Aufbau, Tests, Gem bauen
9. [Ausblick](#9-ausblick)

---

## 1. Überblick

### 1.1 Was das Gem kann

| 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 |
| Benachrichtigung | Berichte an Webhooks (signiert), Slack und E-Mail, je Pipeline oder je Skript; eigene Notifier |
| 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 deren Doku und Tests.

### 1.2 Schnellstart

Im Repository:

```sh
mise install                                          # Ruby 3.4.9 aus mise.toml
bundle install
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
```

In einer anderen Anwendung:

```ruby
# Gemfile
gem 'datapipeline', '~> 3.1', git: 'https://git.wicom.li/wkh/datapipeline.git', branch: 'main'
gem 'pg'                                            # nur die Gems der benötigten Plugins (4.10)
```

`'~> 3.1'` verhindert, dass `bundle update` unbemerkt eine neue Hauptversion mit Breaking Changes holt (8.3). Für die
Entwicklung gegen einen lokalen Checkout: `bundle config set local.datapipeline ../datapipeline`.

```sh
bundle install
bundle exec etl pipeline.rb --dry-run               # das Kommando etl kommt mit dem Gem
```

Die kleinste sinnvolle Pipeline kopiert eine CSV-Datei nach JSON:

```ruby
require 'datapipeline'

DataPipeline.new do
  read  container(:csv, file: 'kunden.csv')
  write container(:json, file: 'kunden.json')
end.run
```

### 1.3 Grundbegriffe

| Begriff | Bedeutung |
| --- | --- |
| 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) |

---

## 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, Benachrichtigung |

### 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
Lauf nicht abbrechen, sondern mit Grund in einer eigenen Datei landen.

```mermaid
flowchart LR
  CSV[kunden.csv] --> V1[validate<br/>Pflicht, Format, Typ]
  LJ[laender.json] -. lookup .-> V2
  V1 --> V2[validate<br/>Land bekannt]
  V2 --> F[filter<br/>nur aktiv]
  F --> M[map<br/>anreichern, umbauen]
  LJ -. lookup .-> M
  M --> DB[(kunden.sqlite3<br/>alle)]
  M -->|only: EU| XL[eu_kunden.xlsx]
  V1 -. on_error .-> ERR[kunden_fehler.csv]
  V2 -. on_error .-> ERR
```

<!-- beispiel: examples/kunden_import.rb -->
```ruby
#
# Kundenliste aus CSV prüfen, mit Ländern anreichern, nach SQLite und Excel schreiben
#
#   bin/etl run examples/kunden_import.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)

DataPipeline.new do
  version 'kunden_import 1.0'

  # Nachschlagetabelle, einmal geladen: { 'DE' => { 'code' => 'DE', 'name' => 'Deutschland', 'eu' => true }, ... }
  laender = lookup container(:json, file: "#{DATA}/laender.json"), key: 'code', unique: true

  read container(:csv, file: "#{DATA}/kunden.csv", csv: { col_sep: ';' })

  # fehlerhafte Zeilen mit Grund in eine eigene Datei, der Lauf geht weiter
  on_error :collect, to: container(:csv, file: "#{out}/kunden_fehler.csv")

  # 1. prüfen
  validate 'kundennr' => { required: true, format: /\A\d{6}\z/ },
           'name' => { required: true, max: 40 },
           'umsatz' => { type: :decimal }
  validate { |row| "Land #{row['land']} unbekannt" unless laender.key?(row['land']) }

  # 2. nur aktive Kunden
  filter { |row| row['status'] == 'aktiv' }

  # 3. anreichern und umbauen
  map do |row|
    land = laender.fetch(row['land'])
    {
      'kundennr' => row['kundennr'],
      'name' => row['name'].strip,
      'land' => land['name'],
      'eu' => land['eu'] ? 'ja' : 'nein',
      'umsatz' => row['umsatz'].to_f.round(2)
    }
  end

  # 4. schreiben: alle Kunden nach SQLite, EU-Kunden zusätzlich als Excel-Liste
  write container(:sqlite, database: "#{out}/kunden.sqlite3", table: 'kunden', create: true, truncate: true)
  write container(:excel, file: "#{out}/eu_kunden.xlsx", sheet: 'EU-Kunden'), only: ->(row) { row['eu'] == 'ja' }

  on_finish { |stats| puts "kunden_import: #{stats}" }
end.run
```

| 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 (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 3.7 |

Ergebnis von `bin/etl run examples/kunden_import.rb`:

```text
INFO start kunden_import 1.0
WARN rejected by validate#1: name: missing
WARN rejected by validate#1: kundennr: format /\A\d{6}\z/
WARN rejected by validate#2: Land XX unbekannt
WARN rejected by validate#1: umsatz: not decimal
INFO finish read=9 written=4 rejected=4 0.097s
kunden_import: read=9 written=4 rejected=4 0.097s
```

| Datei | Inhalt |
| --- | --- |
| `tmp/kunden.sqlite3`, Tabelle `kunden` | 4 aktive, gültige Kunden (DE, AT, CH, IT) mit Ländername, `eu`, `umsatz` als Zahl |
| `tmp/eu_kunden.xlsx`, Blatt `EU-Kunden` | 3 Kunden (ohne CH) |
| `tmp/kunden_fehler.csv` | 4 Zeilen: leerer Name, Kundennummer mit 5 Stellen, Land `XX`, Umsatz `abc` – je mit `_step` und `_error` |

Der Exit-Code ist `2`, weil Zeilen verworfen wurden; ein Abbruch ergäbe `1`, ein fehlerfreier Lauf `0`. Mit
`--dry-run` läuft alles gleich, es entstehen aber weder Datenbank noch Excel- oder Fehlerdatei; stattdessen stehen
die ersten Zeilen auf dem Bildschirm.

### 2.2 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 (Kapitel 5). Ohne `format: true`
formatiert das Layout nicht, deshalb bringt das `map` hier Datum und Beträge selbst ins Zielformat (`delete('-')`,
`format('%.2f', …)`).

<!-- beispiel: examples/rechnungen_satzarten.rb -->
```ruby
#
# Rechnungen aus verschachteltem JSON in eine Datei mit Satzarten (Datei-, Kopf-, Positions-, Schlusssatz)
#
#   bin/etl run examples/rechnungen_satzarten.rb [--out=tmp]
#
require 'fileutils'
require 'datapipeline'

DATA = File.expand_path('data', __dir__)
out = DataPipeline.params.fetch(:out, 'tmp')
FileUtils.mkdir_p(out)

# Feld: [Name, Typ, Länge, Pflicht]; Typ AN Text, N Zahl (11.2 = 9 Vor-, 2 Nachkommastellen), DAT YYYYMMDD
LAYOUT = DataPipeline::Layout.new do
  record 'DATEI', [['SATYP', 'AN', 5, true], ['ERSTELLT', 'DAT', 8, true], ['ANZAHL', 'N', 6, true]]
  record 'KOPF',  [['SATYP', 'AN', 5, true], ['LFDNR', 'N', 6, true], ['BELNR', 'AN', 10, true],
                   ['DATUM', 'DAT', 8, true], ['KUNNR', 'AN', 10, true], ['SUMME', 'N', '11.2', true]]
  record 'POS',   [['SATYP', 'AN', 5, true], ['LFDNR', 'N', 6, true], ['BELNR', 'AN', 10, true],
                   ['POSNR', 'N', 3, true], ['TEXT', 'AN', 30], ['BETRAG', 'N', '11.2', true]]
  record 'ENDE',  [['SATYP', 'AN', 5, true], ['ANZAHL', 'N', 6, true]]
end

DataPipeline.new do
  read container(:json, file: "#{DATA}/rechnungen.json")

  # 1. eine Rechnung -> Kopfsatz + n Positionssätze
  map do |rechnung|
    positionen = rechnung['positionen'].each_with_index.map do |pos, i|
      { 'satyp' => 'POS', 'belnr' => rechnung['nr'], 'posnr' => format('%03d', i + 1),
        'text' => pos['text'], 'betrag' => format('%.2f', pos['menge'] * pos['preis']) }
    end
    kopf = { 'satyp' => 'KOPF', 'belnr' => rechnung['nr'], 'datum' => rechnung['datum'].delete('-'),
             'kunnr' => rechnung['kunde'], 'summe' => format('%.2f', positionen.sum { |p| p['betrag'].to_f }) }
    [kopf, *positionen]
  end

  # 2. laufende Nummer über alle Kopf- und Positionssätze
  map { |row, ctx| row.merge('lfdnr' => format('%06d', ctx.index + 1)) }

  # 3. gegen das Layout prüfen: zu langer Text wird gemeldet, die Zeile bleibt
  validate LAYOUT, warn: true

  # Datei- und Schlusssatz kennen die Anzahl der Datensätze
  header  { |ctx| { 'satyp' => 'DATEI', 'erstellt' => Date.today.strftime('%Y%m%d'), 'anzahl' => format('%06d', ctx.written) } }
  trailer { |ctx| { 'satyp' => 'ENDE', 'anzahl' => format('%06d', ctx.written) } }

  # Felder in der Reihenfolge des Layouts; Positionen zusätzlich als NDJSON
  write container(:csv, file: "#{out}/rechnungen.txt", layout: LAYOUT, csv: { headers: false, col_sep: '|' })
  write container(:json, file: "#{out}/positionen.ndjson"), only: ->(row) { row['satyp'] == 'POS' }
end.run
```

Ergebnis `tmp/rechnungen.txt` (die Warnung meldet den Positionstext mit mehr als 30 Zeichen):

```text
DATEI|20261008|000005
KOPF|000001|R-1001|20251103|100001|380.50
POS|000002|R-1001|001|Beratung|300.00
POS|000003|R-1001|002|Reisekosten|80.50
KOPF|000004|R-1002|20251105|100002|1200.00
POS|000005|R-1002|001|Jahreslizenz Software mit Wartung und Support|1200.00
ENDE|000005
```

### 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.

<!-- beispiel: examples/bewegungen_inkrementell.rb -->
```ruby
#
# Nur neue Buchungsbewegungen exportieren, Zeile für Zeile (Streaming)
# Ein zweiter Lauf ohne neue Daten schreibt nichts.
#
#   bin/etl run examples/bewegungen_inkrementell.rb [--out=tmp]
#
require 'fileutils'
require 'datapipeline'

DATA = File.expand_path('data', __dir__)
out = DataPipeline.params.fetch(:out, 'tmp')
FileUtils.mkdir_p(out)

DataPipeline.new do
  streamed

  # höchste id des letzten erfolgreichen Laufs, gespeichert in tmp/.state.json
  stand = watermark :bewegungen, column: 'id', file: "#{out}/.state.json", initial: 0

  # bei einer Datenbank gehört der Filter ins SQL: where id > $1, params: [stand.value]
  read container(:json, file: "#{DATA}/bewegungen.ndjson")
  filter { |row| row['id'] > stand.value.to_i }
  map { |row| row.merge('soll_haben' => row['betrag'].negative? ? 'H' : 'S', 'betrag' => row['betrag'].abs) }

  write container(:json,
                  file: "#{out}/bewegungen_#{Time.now.strftime('%Y%m%d_%H%M%S')}.ndjson"
                 )

  on_finish { |stats| puts "neu: #{stats.written}, Stand jetzt: #{stand.current}" }
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.

### 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.

<!-- beispiel: examples/eigenes_plugin.rb -->
```ruby
#
# Eigenes Plugin: liest alle CSV-Dateien eines Verzeichnisses als eine Quelle
#
#   bin/etl run examples/eigenes_plugin.rb [--out=tmp]
#
require 'fileutils'
require 'datapipeline'

DATA = File.expand_path('data', __dir__)
out = DataPipeline.params.fetch(:out, 'tmp')
FileUtils.mkdir_p(out)

class VerzeichnisStorage < DataPipeline::StoragePlugin
  def initialize dir:, pattern: '*.csv', **csv_options
    @dir = dir
    @pattern = pattern
    @csv_options = csv_options
  end

  # jede Zeile bekommt den Dateinamen dazu; mit Block zeilenweise (für streamed)
  def read(&block)
    rows = Dir.glob(File.join(@dir, @pattern)).sort.flat_map do |file|
      DataPipeline::CsvStorage.new(file:, **@csv_options).read.map { |row| row.merge('datei' => File.basename(file, '.*')) }
    end
    return rows unless block

    rows.each(&block)
    nil
  end
end

DataPipeline.register :verzeichnis, VerzeichnisStorage

DataPipeline.new do
  read container(:verzeichnis, dir: "#{DATA}/filialen")
  map { |row| row.merge('menge' => row['menge'].to_i) }
  write container(:csv, file: "#{out}/alle_filialen.csv", columns: %w[datei artikel menge])
end.run
```

Ergebnis `tmp/alle_filialen.csv`:

```text
datei,artikel,menge
vaduz,A-100,5
vaduz,B-200,2
zuerich,A-100,7
zuerich,C-300,1
```

### 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
und antwortet einmal absichtlich mit `503`, damit die Wiederholung sichtbar wird.

```sh
ruby examples/demo_api.rb &                  # http://127.0.0.1:4567, Token demo-token
bin/etl run examples/rest_lesen.rb
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 Umgebung).

| Endpunkt der Demo-API | Antwort |
| --- | --- |
| `GET /api/kunden?page=1&per_page=3` | `{"data": [...], "page": 1}`; Seite 2 antwortet einmal mit `503` |
| `GET /api/laender` | `[{"code": "DE", "name": "Deutschland"}, ...]` |
| `GET /api/bestellungen?cursor=n` | `{"items": [...], "next": "/api/bestellungen?cursor=n+2"}`, auf der letzten Seite `"next": null` |
| `POST /api/import` | ein Objekt oder Array; `422`, wenn `kundennr` oder `name` fehlt |
| `PUT /api/archiv/<id>` | legt das Objekt unter der id ab |

**Lesen: seitenweise, mit Wiederholung und Nachschlagen** – `page_param` holt Seite für Seite, bis eine leer ist;
`root` zeigt auf das Array in der Antwort. Ein zweiter Container auf denselben Server dient mit `query` als Lookup.

<!-- beispiel: examples/rest_lesen.rb -->
```ruby
#
# Kunden seitenweise aus einer REST-API lesen, mit Ländern aus einem zweiten Endpunkt anreichern,
# nach CSV und JSON schreiben
#
#   ruby examples/demo_api.rb &                         # Demo-API starten
#   bin/etl run examples/rest_lesen.rb [--url=http://127.0.0.1:4567] [--out=tmp]
#
require 'fileutils'
require 'datapipeline'

API = DataPipeline.params.fetch(:url, 'http://127.0.0.1:4567')
AUTH = "Bearer #{ENV.fetch('DEMO_API_TOKEN', 'demo-token')}"
out = DataPipeline.params.fetch(:out, 'tmp')
FileUtils.mkdir_p(out)

DataPipeline.new do
  version 'rest_lesen 1.0'

  # GET /api/kunden?page=1&per_page=3, dann page=2, ... bis eine Seite leer ist
  # die Zeilen stehen in der Antwort unter "data"; 429/5xx und Netzfehler werden bis zu 3-mal wiederholt
  kunden = container :rest, url: "#{API}/api/kunden", auth: AUTH,
                            root: 'data', page_param: 'page', params: { 'per_page' => 3 },
                            retries: 3, retry_wait: 0.2

  # derselbe Server, anderer Endpunkt: query hängt den Pfad an die url an (GET /api/laender)
  stamm = container :rest, url: "#{API}/api", auth: AUTH
  laender = lookup stamm, query: 'laender', key: 'code', unique: true

  read kunden

  on_error :collect, to: container(:csv, file: "#{out}/api_fehler.csv")
  validate 'id' => { required: true }, 'umsatz' => { type: :decimal }

  map do |kunde|
    {
      'id' => kunde['id'],
      'name' => kunde['name'],
      'land' => laender.dig(kunde['land'], 'name') || kunde['land'],
      'umsatz' => kunde['umsatz'].to_f
    }
  end

  write container(:csv, file: "#{out}/api_kunden.csv", columns: %w[id name land umsatz])
  write container(:json, file: "#{out}/api_kunden.json", pretty: true)

  on_finish { |stats| puts "rest_lesen: #{stats}" }
end.run
```

```text
INFO start rest_lesen 1.0
WARN rejected by validate#1: umsatz: not decimal
INFO finish read=7 written=6 rejected=1 0.213s
```

Gelesen werden 7 Kunden in 3 Seiten (plus eine leere Seite als Ende); die `503` auf Seite 2 bemerkt man nur an der
Laufzeit. Kunde 5 hat den Umsatz `n/a` und steht in `tmp/api_fehler.csv`. Ergebnis `tmp/api_kunden.csv`:

```text
id,name,land,umsatz
1,Müller GmbH,Deutschland,12500.5
2,Huber & Partner,Österreich,8300.0
3,Swiss Trade AG,Schweiz,22000.75
4,Rossi Srl,Italien,4100.1
6,Nordic AB,SE,990.0
7,Vaduz Treuhand,Liechtenstein,15000.0
```

Schweden fehlt in `/api/laender`, deshalb bleibt bei Kunde 6 der Code `SE` stehen.

**Schreiben: prüfen, dann in Paketen senden** – `batch_size: 2` schickt je 2 Zeilen als JSON-Array. Was die API
ablehnen würde, wird vorher geprüft: Lehnt die API ein Paket trotzdem ab (`4xx`), bricht der Lauf ab, weil ein Fehler
beim Schreiben nicht einer einzelnen Zeile zugeordnet werden kann.

<!-- beispiel: examples/rest_schreiben.rb -->
```ruby
#
# Kunden aus CSV prüfen und in Paketen per POST an eine REST-API senden
#
#   ruby examples/demo_api.rb &
#   bin/etl run examples/rest_schreiben.rb [--url=http://127.0.0.1:4567] [--dry-run]
#
require 'fileutils'
require 'datapipeline'

API = DataPipeline.params.fetch(:url, 'http://127.0.0.1:4567')
AUTH = "Bearer #{ENV.fetch('DEMO_API_TOKEN', 'demo-token')}"
DATA = File.expand_path('data', __dir__)
out = DataPipeline.params.fetch(:out, 'tmp')
FileUtils.mkdir_p(out)

DataPipeline.new do
  version 'rest_schreiben 1.0'

  read container(:csv, file: "#{DATA}/kunden.csv", csv: { col_sep: ';' })

  # vor dem Senden prüfen: was die API ablehnen würde, landet hier in der Fehlerdatei
  on_error :collect, to: container(:csv, file: "#{out}/import_fehler.csv")
  validate 'kundennr' => { required: true, format: /\A\d{6}\z/ }, 'name' => { required: true }

  filter { |row| row['status'] == 'aktiv' }
  map do |row|
    { 'kundennr' => row['kundennr'], 'name' => row['name'].strip, 'land' => row['land'], 'umsatz' => row['umsatz'].to_f }
  end

  # POST /api/import mit je 2 Kunden als JSON-Array; im Dry-Run wird nichts gesendet
  write container(:rest, url: "#{API}/api/import", auth: AUTH, batch_size: 2)

  on_finish do |stats|
    puts stats.dry_run ? 'Dry-Run: nichts gesendet' : "gesendet: #{stats.written} Kunden in #{(stats.written / 2.0).ceil} Paketen"
  end
end.run
```

Ergebnis: 6 aktive Kunden in 3 Anfragen an `POST /api/import`, 2 Kunden in `tmp/import_fehler.csv` (leerer Name,
Kundennummer mit 5 Stellen). Mit `--dry-run` wird nichts gesendet; auf dem Bildschirm stehen die Daten, die gesendet
würden:

```text
DRY RUN – nothing written: read=9 written=6 rejected=2 0.0s (dry run)
{"kundennr":"100001","name":"Müller GmbH","land":"DE","umsatz":12500.5}
...
```

**Von API zu API: Folge-Links, Streaming, eine URL je Datensatz** – Die Antwort nennt die nächste Seite selbst; das
Lambda in `next:` macht daraus eine vollständige URL. `write_url` als Lambda schickt jede Bestellung an ihre eigene
Adresse. Mit dem Wasserzeichen archiviert ein zweiter Lauf nur, was neu dazugekommen ist.

<!-- beispiel: examples/rest_zu_rest.rb -->
```ruby
#
# Von API zu API: neue Bestellungen über Folge-Links lesen und einzeln per PUT archivieren
# Ein zweiter Lauf ohne neue Bestellungen sendet nichts.
#
#   ruby examples/demo_api.rb &
#   bin/etl run examples/rest_zu_rest.rb [--url=http://127.0.0.1:4567] [--out=tmp]
#
require 'fileutils'
require 'datapipeline'

API = DataPipeline.params.fetch(:url, 'http://127.0.0.1:4567')
AUTH = "Bearer #{ENV.fetch('DEMO_API_TOKEN', 'demo-token')}"
out = DataPipeline.params.fetch(:out, 'tmp')
FileUtils.mkdir_p(out)

DataPipeline.new do
  streamed

  # höchste Bestellnummer des letzten erfolgreichen Laufs
  stand = watermark :bestellungen, column: 'id', file: "#{out}/.state.json", initial: 0

  # die Antwort nennt die nächste Seite selbst: {"items": [...], "next": "/api/bestellungen?cursor=2"}
  read container(:rest, url: "#{API}/api/bestellungen", auth: AUTH, root: 'items',
                        next: ->(body) { body['next'] && "#{API}#{body['next']}" })

  filter { |row| row['id'] > stand.value.to_i }
  map { |row| row.merge('archiviert' => Date.today.iso8601) }

  # eine url je Bestellung: PUT /api/archiv/<id>
  write container(:rest, url: "#{API}/api/archiv", auth: AUTH, write_method: :put,
                         write_url: ->(row) { "#{API}/api/archiv/#{row['id']}" })

  on_finish { |stats| puts "archiviert: #{stats.written}, Stand jetzt: #{stand.current}" }
end.run
```

```text
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]
```

<!-- beispiel: examples/buchungen_export.rb -->
```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__)

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
  notify DataPipeline::Slack.new(channel: '#etl') if ENV['SLACK_TOKEN']   # Meldung nur bei Problemen

  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 { |stats| puts "#{stats.name}: #{stats}" }
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) |
| `notify DataPipeline::Slack…` | 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`.)

---

## 3. Pipelines schreiben

### 3.1 Ablauf eines Laufs

```mermaid
flowchart LR
  SRC((Quelle)) --> STEPS[Schritte<br/>validate · filter · map<br/>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))
  EDGE --> ST[Stats · history · on_finish]
```

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.

### 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 |
| --- | --- |
| `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 |
| `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 |
| `notify(notifier, on: :problems)`, `notifier(:name)` | Bericht an Webhook, Slack oder E-Mail (6.8) |
| `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` |
| `configure(opts)` | mehrere Einstellungen als Hash, z. B. `configure(streamed: true, version: '1.0')` |
| `run` | führt die Pipeline aus und liefert `DataPipeline::Stats` |
| `DataPipeline.new(**settings) { … }` | Einstellungen nur für diese Pipeline, z. B. `new(limit: 10)` |
| `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) |

### 3.3 Schritte

**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.

**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).

```ruby
map { |row, ctx| row.merge('lfdnr' => format('%06d', ctx.index + 1)) }
trailer { |ctx| { 'satyp' => 'ENDE', 'anzahl' => ctx.written } }
```

**Bedingung** – `condition: ->(row) { … }` an `map` oder `filter` wendet den Schritt nur auf passende Zeilen an; die
anderen fallen in diesem Schritt weg.

**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.

**Ohne Schritte** wird die Quelle unverändert in die Ziele kopiert.

**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` |
| `reconcile` | vor dem Schreiben | nach der letzten Zeile; die Zeilen sind schon geschrieben |
| Postgres, SQLite als Ziel | eine Transaktion für alles | eine Transaktion je `batch_size` Zeilen, mit `batch_size: :all` eine für alles (4.2, 4.3) |
| Postgres als Quelle | eine Abfrage | serverseitiger Cursor, je `fetch_size` Zeilen |

`limit n` beendet im Streaming das Lesen nach n Zeilen.

### 3.4 Kurzschritte

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 `<art>#<nummer>` 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) 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 oder Datei vorbereiten; möglichst erst bei der ersten Nutzung öffnen (Dry-Run) |
| `read` → `Array<Hash<String, …>>` | 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<Hash>` | 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
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 `<Name>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]
write container(:postgres, connection: :crm, table: 'beleg', upsert: %w[belnr], batch_size: :all)
```

| 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(:sqlite, database: 'tmp/kunden.sqlite3', table: 'kunden', create: true, truncate: true)
```

| 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 |
| `batch_size` | Streaming: Zeilen je Transaktion (Default 1000) oder `:all` |

Werte kommen mit SQLite-Typen zurück (Integer, Float, String, `nil`); `Date` und `Time` werden als ISO-Text
geschrieben. `write` schreibt in einer Transaktion. Im Streaming gilt dasselbe wie bei Postgres (4.2): `write_row`
schreibt je `batch_size` Zeilen in einer Transaktion, `finalize` und `close` den Rest; mit `batch_size: :all` ist der
ganze Stream eine Transaktion, die nach einem Abbruch zurückgerollt wird. Eine Abfrage (`query`, `lookup`) auf
demselben Storage schreibt noch gesammelte Zeilen vorher, sie sieht also alles, was schon übergeben wurde.

### 4.4 CSV

```ruby
read  container(:csv, file: 'kunden.csv', csv: { col_sep: ';' })
write container(:csv, file: 'export.csv', columns: %w[id name land], encoding: 'Windows-1252')
```

| 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 |

- 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.

### 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', 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
select %w[id name land]
write xlsx
```

### 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:

| Plugin | Gem |
| --- | --- |
| `:postgres` | `pg` |
| `:sqlite` | `sqlite3` |
| `:excel` | `roo` (lesen), `caxlsx` (schreiben) |
| `:rest`, `DataPipeline::Slack`, `DataPipeline::Webhook` | `http` |
| `DataPipeline::Email` | `net-smtp` |
| `:csv`, `:json`, `:fixed` | – |

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.

---

## 5. Satzarten-Layouts

### 5.1 Wozu

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 |
| --- | --- |
| 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 |

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
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"]
```

### 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
read container(:csv, file: 'in/cudo.ok', layout: LAYOUT, csv: { headers: false, col_sep: '|' })
# { 'satyp' => 'LARC', 'summe' => '12' }
```

### 5.5 Festbreiten-Dateien

`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 |
| --- | --- |
| `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) |

Eine Satzart, die nicht im Layout steht, ist immer ein Fehler (`record type "X" not in layout (DOHD, LARC)`).

---

## 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
```

| 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) |
| `--notify=NAME`, `--notify-on=WHEN` | ein Bericht für alle Pipelines des Skripts an den Notifier `NAME` aus der Verbindungsdatei; `WHEN` = `problems` (Default), `failure`, `always` (6.8) |
| `--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 Benachrichtigungen

`notify` schickt nach einem Lauf einen Bericht an einen Notifier: einen Webhook, Slack oder E-Mail. Ein Fehler beim
Senden wird nur geloggt; Ergebnis und Exit-Code bleiben unverändert.

```ruby
DataPipeline.new do
  name 'cudo_export'
  notify DataPipeline::Webhook.new(url: ENV['ALERT_URL'], secret: ENV['ALERT_SECRET'])   # bei Problemen
  notify DataPipeline::Slack.new(channel: '#etl', mention: '<!here>'), on: :failure
  notify DataPipeline::Email.new(to: 'buchhaltung@example.com'), on: :always
  notify notifier(:alerts)                                                               # aus der Verbindungsdatei
  # … read, Schritte, write …
end.run
```

| Option von `notify` | Wirkung |
| --- | --- |
| `on: :problems` (Default) | bei Abbruch, verworfenen Zeilen oder Warnungen |
| `on: :failure` | nur bei Abbruch |
| `on: :always` | nach jedem Lauf |
| `title:` | Titel des Berichts, Default der Name der Pipeline |
| `dry_run: true` | auch nach einem Dry-Run senden (Default: nicht) |
| `include_rows: true` | die verworfenen Zeilen mitsenden; Default: nur Schritt und Meldung, weil Zeilen Kundendaten enthalten können |

**Bericht** – Jeder Notifier bekommt einen `DataPipeline::Report` mit den Stats eines oder mehrerer Läufe:

| Methode | Inhalt |
| --- | --- |
| `event` | `run.failed` (Abbruch), `run.rejected` (verworfene Zeilen), `run.warnings`, `run.succeeded`; bei mehreren Läufen der schlimmste |
| `title`, `status`, `subject` | z. B. `cudo_export`, `mit verworfenen Zeilen`, `cudo_export: mit verworfenen Zeilen` |
| `lines`, `to_s` | Text je Lauf: Zähler, Start, Dauer, die ersten 5 Fehler (ohne Zeilen) |
| `to_h(include_rows: false)` | JSON-Payload für Webhooks (unten) |
| `id`, `host`, `params`, `runs`, `error` | eindeutige Kennung, Rechner, Parameter der Kommandozeile, Stats, Fehler ausserhalb eines Laufs |

**Ein Bericht für das ganze Skript** – `notify` meldet jede Pipeline einzeln. Für einen Sammelbericht über alle
Pipelines eines Skripts gibt es die Kommandozeile; der Notifier kommt aus der Verbindungsdatei:

```sh
bin/etl run cudo.rb --notify=alerts                     # bei Problemen (Default)
bin/etl run cudo.rb --notify=alerts --notify-on=always
```

Bricht das Skript selbst ab (z. B. ein Fehler vor der ersten Pipeline), meldet der Bericht `run.failed` mit diesem
Fehler. Bei belegter Sperre (Exit-Code `3`) und im Dry-Run wird nichts gesendet.

#### Webhook

`DataPipeline::Webhook` sendet den Bericht als JSON per `POST`.

| Option | Default | Bedeutung |
| --- | --- | --- |
| `url:` | – | Ziel, Pflicht |
| `secret:` | – | signiert den Body (unten) |
| `headers:` | `{}` | weitere Header, z. B. `{ 'Authorization' => "Bearer #{ENV['TOKEN']}" }` |
| `timeout:` | 10 | Sekunden je Anfrage |
| `retries:`, `retry_wait:` | 3, 2 | Wiederholung bei 429, 5xx und Netzwerkfehlern, Wartezeit `retry_wait` × Versuch; `Retry-After` wird beachtet; andere 4xx brechen sofort ab |

```json
{"schema":1,"id":"3f6c2a1e-…","event":"run.rejected","title":"cudo_export","host":"etl01",
 "params":{"from":"2026-10-08"},"error":null,
 "runs":[{"name":"cudo_export","version":"1.0","started_at":"2026-10-09T06:00:00.123+02:00","duration":4.213,
          "read":1250,"written":1243,"rejected":7,"warnings":0,"ok":false,"error":null,
          "totals":{"dmbtr":{"read":"98231.4","written":"97810.1"}},
          "errors":[{"step":"validate#1","error":"konto: missing"}]}]}
```

`schema: 1` ist ein Versprechen: Felder können dazukommen, eine Änderung bestehender Felder hiesse `schema: 2`. Jede
Anfrage trägt die Header `X-DataPipeline-Event` und `X-DataPipeline-Delivery` (die `id`, bei Wiederholungen gleich;
damit erkennt der Empfänger Doppelte). Mit `secret:` kommen dazu:

```text
X-DataPipeline-Timestamp: 1791525600
X-DataPipeline-Signature: sha256=<HMAC-SHA256(secret, "<timestamp>.<body>") hexadezimal>
```

Der Empfänger prüft Signatur und Alter (gegen wiederholt eingespielte Anfragen); in Ruby z. B.:

```ruby
ok = DataPipeline::Webhook.valid_signature?(secret: ENV['ALERT_SECRET'], body: request.raw_post,
                                            timestamp: request.headers['X-DataPipeline-Timestamp'],
                                            signature: request.headers['X-DataPipeline-Signature'],
                                            max_age: 300)
```

#### Slack

`DataPipeline::Slack` meldet einen Bericht als Statuszeile mit Symbol (🔴 abgebrochen, 🟠 verworfene Zeilen,
🟡 Warnungen, 🟢 ok) und darunter den Text je Lauf. `&`, `<` und `>` aus Daten und Meldungen werden maskiert.

Zwei Wege zu Slack:

| Weg | Einrichten | Aufruf |
| --- | --- | --- |
| Incoming Webhook (einfacher) | Slack-App anlegen (api.slack.com/apps), *Incoming Webhooks* aktivieren, Webhook für den Channel erzeugen | `Slack.new(webhook: ENV['SLACK_WEBHOOK'])` |
| Bot-Token | Slack-App mit Bot-Scope `chat:write` installieren, *Bot User OAuth Token* (`xoxb-…`) kopieren, App in den Channel einladen (`/invite @appname`) | `Slack.new(channel: '#etl')`, Token aus `ENV['SLACK_TOKEN']` |

| Option | Default | Bedeutung |
| --- | --- | --- |
| `webhook:` | – | URL eines Incoming Webhook; dann braucht es weder Token noch Channel |
| `token:` | `ENV['SLACK_TOKEN']` | Bot-Token; ohne Token und ohne Webhook bricht schon `Slack.new` mit `ArgumentError` ab |
| `channel:` | `'#general'` | Channel als Name oder ID (nur mit Token) |
| `mention:` | – | z. B. `'<!here>'` oder `'<!subteam^ID>'`, nur bei `run.failed` |
| `timeout:`, `retries:`, `retry_wait:` | 10, 3, 2 | wie beim Webhook |

Eigener Text geht weiter mit `send_message`, z. B. in `on_finish`:

```ruby
slack = DataPipeline::Slack.new(channel: '#etl')
slack.send_message('Monatsabschluss exportiert', verbose: true)    # verbose: Antwort von Slack auf $stdout
```

Meldet Slack einen Fehler (`invalid_auth`, `channel_not_found`, `not_in_channel`), wirft der Aufruf
`DataPipeline::Slack::Error` mit diesem Code.

#### E-Mail

`DataPipeline::Email` sendet den Bericht als Text-Mail (UTF-8) über SMTP. Betreff: `[etl] <Titel>: <Status>`. Die
Zugangsdaten kommen aus der Umgebung oder als Option:

| Option | Default | Bedeutung |
| --- | --- | --- |
| `to:` | – | eine Adresse oder ein Array, Pflicht |
| `from:` | `ENV['SMTP_FROM']`, sonst `etl@<rechner>` | Absender |
| `host:`, `port:` | `ENV['SMTP_HOST']` bzw. `localhost`; `ENV['SMTP_PORT']` bzw. 587 | SMTP-Server |
| `username:`, `password:` | `ENV['SMTP_USERNAME']`, `ENV['SMTP_PASSWORD']` | Anmeldung (LOGIN); ohne Benutzer keine Anmeldung |
| `starttls:` | `:auto` | `:auto` (wenn der Server es anbietet), `:always` oder `false` (dann Default-Port 25) |
| `tls:` | `false` | `true` = TLS von Anfang an (Port 465) |
| `subject_prefix:` | `'[etl] '` | Anfang des Betreffs |
| `timeout:` | 15 | Sekunden für Verbindung und Antwort |

Mit `include_rows: true` stehen die verworfenen Zeilen als JSON am Ende der Mail. Eigener Text:
`DataPipeline::Email.new(to: …).send_mail('Betreff', "Text")`. Ein Fehler beim Senden wirft
`DataPipeline::Email::Error`. Das Plugin braucht das Gem `net-smtp`.

#### Notifier in der Verbindungsdatei

Wie Storages lassen sich Notifier in `config/connections.yml` ablegen (6.2); `type` ist `webhook`, `slack` oder
`email`, die anderen Schlüssel sind die Optionen:

```yaml
alerts:
  type: webhook
  url: <%= ENV['ALERT_URL'] %>
  secret: <%= ENV['ALERT_SECRET'] %>
buchhaltung_mail:
  type: email
  to: [buchhaltung@example.com, etl@example.com]
  host: smtp.example.com
  username: etl@example.com
  password: <%= ENV['SMTP_PASSWORD'] %>
```

```ruby
notify notifier(:buchhaltung_mail), on: :always
```

#### Eigene Notifier

Ein Notifier ist eine Klasse mit Basisklasse `DataPipeline::Notifier` und einer Methode
`deliver(report, include_rows: false)`, z. B. für Microsoft Teams oder ein Ticketsystem:

```ruby
require 'http'

class TeamsNotifier < DataPipeline::Notifier
  def initialize(url:) = @url = url

  def deliver report, include_rows: false
    HTTP.post(@url, json: { title: report.subject, text: report.lines.join("\n\n") })
  end
end

notify TeamsNotifier.new(url: ENV['TEAMS_WEBHOOK']), on: :failure
```

### 6.9 Fehlersuche

| Meldung | Ursache | Abhilfe |
| --- | --- | --- |
| `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 |
| `NoMethodError: undefined method 'transform'` (oder `chained`) | Skript für 2.x | auf `map`, `header`, `trailer` umstellen (7.1) |
| `map: condition must be a lambda ->(row) { ... }` | `condition:` als String aus 2.x | `condition: ->(row) { … }` (7.1) |

Mehr sehen: `--verbose` (Log-Level Debug), `verbose` in der Pipeline (Daten als JSON), `--dry-run --limit=20` für einen
schnellen Blick auf das Ergebnis.

---

## 7. Umstieg

### 7.0 Von 3.0 auf 3.1 und 3.2

3.1 und 3.2 sind abwärtskompatibel. Anders als in 3.0:

| Thema | bis 3.0 | jetzt | seit |
| --- | --- | --- | --- |
| SQLite, `write_row` im Streaming | jede Zeile einzeln | je 1000 Zeilen in einer Transaktion (4.3); eine zweite Verbindung sieht die Zeilen paketweise | 3.1 |
| Lizenz | nicht festgelegt | BSD-3-Clause | 3.1 |
| Slack-Fehler | `RuntimeError` | `DataPipeline::Slack::Error` (weiterhin ein `StandardError`) | 3.2 |
| `Slack.new` | lädt `http` erst beim Senden | lädt `http` sofort; Timeout 10 s und Wiederholung bei 429/5xx | 3.2 |
| `send_message(…, verbose: true)` | Antwort als eingerücktes JSON | Antwort wie von Slack geliefert | 3.2 |

Selbstgebaute Slack-Meldungen in `on_finish` laufen weiter; einfacher ist `notify` (6.8).

### 7.1 Von 2.x auf 3.0

3.0 entfernt das alte Ausführungsmodell: `transform`, den Parallelmodus, `chained` und `condition:` als String
(`eval`). Alle Schritte laufen jetzt verkettet, jeder bekommt die Ausgabe des vorigen. Alles andere ist unverändert;
ein Skript, das unter 2.1 ohne `WARN deprecated` lief, läuft auch unter 3.0.

| 2.x | 3.0 |
| --- | --- |
| `transform { \|row, index\| … }` | `map { \|row, ctx\| … }`, Index als `ctx.index` |
| `transform(processing: :prepend) { \|_, _\| kopf }` | `header { \|ctx\| kopf }` |
| `transform(processing: :append) { \|_, n\| schluss }` | `trailer { \|ctx\| schluss }`, Anzahl als `ctx.written` |
| `transform(condition: "row['a'] == 1") { … }` | `map(condition: ->(row) { row['a'] == 1 }) { … }` |
| `chained`, `configure(chained: true)` | weglassen |

Worauf beim Umstellen zu achten ist:

- **Parallelmodus:** Ohne `map`, `filter`, `validate` und `chained` bekam unter 2.x jeder `transform` eine eigene
  Kopie der Quellzeilen, und die Ergebnisse wurden hintereinander angehängt. Zwei `transform` auf 2 Quellzeilen
  ergaben 4 Zeilen, zwei `map` ergeben 2 Zeilen mit beiden Änderungen. Wer wirklich jede Zeile mehrfach ausgeben will,
  gibt in einem `map` ein Array zurück: `map { |r| [mit_b(r), mit_c(r)] }`.
- **Kopfsatz:** `transform(processing: :prepend)` lief vor den Daten, `header` läuft im Batch erst danach (steht aber
  vorne). Das zählt nur, wenn der Block etwas anderes tut, als eine Zeile zu liefern.
- **Schlusssatz:** `processing: :append` bekam im Parallelmodus die Anzahl der Quellzeilen; `ctx.written` ist die
  Anzahl der Datenzeilen nach allen Schritten, `ctx.read` die der Quellzeilen.
- **Bedingung:** Ein String in `condition:` bricht schon beim Aufbau der Pipeline mit `ArgumentError` ab.

Prüfen lässt sich die Umstellung mit `etl run skript.rb --dry-run` vorher und nachher: Die Zusammenfassung und die
ersten Zeilen sollten gleich sein.

### 7.2 Von 2.0 auf 2.1

2.1 war abwärtskompatibel. Anders als in 2.0:

| Thema | 2.0 | ab 2.1 |
| --- | --- | --- |
| Postgres, `write_row` im Streaming | jede Zeile sofort, ohne Transaktion | je 1000 Zeilen in einer Transaktion; der Rest in `finalize`/`close`, auch nach einem Abbruch |
| CSV lesen mit UTF-8-BOM | BOM steht im ersten Spaltennamen | BOM wird übersprungen |
| `Stats#to_s` | – | mit `totals` zusätzlich `spalte=quelle/ergebnis` |
| `etl` ohne `name` | – | `stats.name` = Skriptname |

---

## 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/](lib/datapipeline/notification/) | `notifier.rb` (Report, Basisklasse), `webhook.rb`, `slack.rb`, `email.rb` |
| [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 210 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, Webhook und Slack gegen einen lokalen HTTP-Testserver, E-Mail gegen einen SMTP-Testserver (`test/support/`); Laden der Gems erst bei Nutzung |
| `test/core` | DSL, Schritte und Kurzschritte, Batch und Streaming, Fehlerbehandlung, Layouts, Sperre, Verbindungen, Summen, Historie, Wasserzeichen |
| `test/cli` | `bin/etl` als Prozess: Parameter, Dry-Run, Sperre, Historie, Sammelbericht `--notify`, 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 (wie 3.0, 7.1).
- Jede Version bekommt ein Tag `vX.Y.Z` (`git tag -a v3.2.0 -m "datapipeline 3.2.0" && git push origin v3.2.0`).
  Anwendungen binden das Gem mit einer Versionsgrenze ein (`'~> 3.1'`, 1.2) oder fest auf ein Tag
  (`tag: 'v3.2.0'`; dann geht `local.datapipeline` nicht).
- Das `Gemfile` bindet das Gem über `gemspec` ein und ergänzt die Gruppen `plugins` (`pg`, `sqlite3`, `roo`, `net-smtp`,
  `caxlsx`, `http`) und `test`; `Gemfile.lock` ist versioniert. Ohne Plugin-Gems: `bundle config set without plugins`.
- Gem bauen: `gem build datapipeline.gemspec`.
- Lizenz: BSD-3-Clause, siehe [LICENSE](LICENSE); die Datei ist Teil des Gems.
- Beispiele in Kapitel 2 werden mit einem Marker `<!-- beispiel: examples/datei.rb -->` 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. **Benachrichtigungen:** Microsoft Teams als mitgelieferter Notifier, HTML-Mail.
4. **Weitere Plugins:** SFTP-Ablage.