Custom Airflow Library der Datenfabrik

Datenfabrik Libraries — Installation

Das Paket datenfabrik-libraries enthält die plattformspezifischen Hooks und Hilfsfunktionen der Datenfabrik. Es kann über die GitLab Package Registry bezogen werden und ist öffentlich verfügbar. Es ist auch bereits in dem von der Datenfabrik verwendeten airflow image installiert.

pip

pip install datenfabrik-libraries \
  --index-url https://gitlab.opencode.de/api/v4/projects/12819/packages/pypi/simple

# Mit Airflow dependencies
pip install "datenfabrik-libraries[airflow]" \
  --index-url https://gitlab.opencode.de/api/v4/projects/12819/packages/pypi/simple

In einer requirements.txt

--index-url https://gitlab.opencode.de/api/v4/projects/12819/packages/pypi/simple
datenfabrik-libraries[airflow]

Das [airflow]-Extra bezieht apache-airflow und apache-airflow-providers-amazon als Abhängigkeiten.

DF_S3Hook

Überblick

Der DF_S3Hook ist ein Wrapper um den Standard-Airflow S3Hook, der die Verwendung von Lifecycle-Tags für S3-Operationen verpflichtend macht. Er bietet keine zusätzlichen Features, sondern erzwingt lediglich eine konsistente Datenkategorisierung durch das data-type Tag.

Was ist der Standard S3Hook?

Der Standard-Airflow S3Hook ermöglicht die Interaktion mit S3 (Datenfabrik S3 Lösung):

# ❌ Standard Airflow S3Hook - NICHT in der Datenfabrik verwenden!
from airflow.providers.amazon.aws.hooks.s3 import S3Hook

hook = S3Hook(aws_conn_id='S3-MINIO')
hook.load_file(
    filename='/tmp/data.csv',
    bucket_name='my-bucket',
    key='data/file.csv'
)
# Problem: Keine Lifecycle-Kategorisierung, nicht erwünscht.

Zweck des Datenfabrik S3 Hook

  • Verhindert, dass Daten ohne Lifecycle-Kategorisierung in S3 geschrieben werden
  • Stellt sicher, dass alle S3-Objekte mit einem data-type Tag versehen sind

Verpflichtender Parameter

Der Hook erfordert bei der Initialisierung zwingend den extra_args Parameter mit einem data-type Tag:

extra_args={'Tagging': 'data-type=<value>'}

Erlaubte data-type Werte

Wert Beschreibung Verwendung
raw Rohdaten aus externen Quellen Daten direkt nach dem Download/Import von externen APIs, Datenbanken oder Dateien
intermediate Verarbeitete/transformierte Daten Zwischenergebnisse während der Datenverarbeitung, bereinigte oder angereicherte Daten
output Finale Ausgabedaten Endgültige Daten, die für den Konsum durch Endnutzer oder andere Systeme bereit sind

Verwendung

Basis-Beispiel

from datenfabrik_libraries.hooks import DF_S3Hook

# Hook mit data-type Tag initialisieren
s3hook = DF_S3Hook(
    aws_conn_id='S3-MINIO',
    extra_args={'Tagging': 'data-type=raw'}
)

# Datei hochladen ("raw" Tag wird automatisch angewendet)
s3hook.load_file(
    filename='/tmp/data.csv',
    bucket_name='my-bucket',
    key='raw/data.csv'
)

Verschiedene Data-Types in einem DAG

from datenfabrik_libraries.hooks import DF_S3Hook

# Hook für Rohdaten
hook_raw = DF_S3Hook(
    aws_conn_id='S3-MINIO',
    extra_args={'Tagging': 'data-type=raw'}
)

# Hook für Zwischenergebnisse
hook_intermediate = DF_S3Hook(
    aws_conn_id='S3-MINIO',
    extra_args={'Tagging': 'data-type=intermediate'}
)

# Hook für finale Ausgaben
hook_output = DF_S3Hook(
    aws_conn_id='S3-MINIO',
    extra_args={'Tagging': 'data-type=output'}
)

Fehlerbehandlung

Wenn das data-type Tag fehlt oder einen ungültigen Wert hat, wird eine DF_DataTypeTagMissingError Exception geworfen.

from datenfabrik_libraries.hooks import DF_S3Hook, DF_DataTypeTagMissingError

Verfügbare Methoden

Da der DF_S3Hook vom Standard-S3Hook erbt, stehen alle Standard-S3-Operationen zur Verfügung.

FileAccess / FileAccessS3

Überblick

datenfabrik_libraries.helpers bietet zwei Hilfsklassen für einheitlichen Dateizugriff — unabhängig davon, ob die Daten lokal oder in S3 liegen:

Klasse Backend Beschreibung
FileAccess Lokales Dateisystem Lesen/Schreiben von Dateien in einem festen Basisverzeichnis
FileAccessS3 S3 / MinIO Lesen/Schreiben von Dateien in einem S3-Bucket (via S3Hook)

Beide Klassen implementieren dieselbe API, sodass Code ohne Änderungen für beide Storage-Methoden verwendet werden kann.

Gemeinsame API

Methode Beschreibung
read(filename) Dateiinhalt als String lesen
read_json(filename) JSON-Datei lesen und als Objekt zurückgeben
write(data, filename) Binärdaten schreiben/hochladen
csvwriter(filename) Context Manager: CSV-Writer (;-getrennt)
pathbuf(filename) Pfad/Buffer für lesende Bibliothekszugriffe (Text)
pathbuf_b(filename) Buffer für lesende Bibliothekszugriffe (Bytes)
pathbuf_w(filename) Context Manager: schreibbarer BytesIO-Buffer, wird beim Verlassen hochgeladen

Verwendung

Lokaler Dateizugriff

from datenfabrik_libraries.helpers import FileAccess

storage = FileAccess("/data/my-dir")

# Lesen
content = storage.read("input.csv")
data = storage.read_json("config.json")

# Schreiben (CSV)
with storage.csvwriter("output.csv") as writer:
    writer.writerow(["col1", "col2"])
    writer.writerow(["value1", "value2"])

S3-Dateizugriff im Airflow DAG

from datenfabrik_libraries.hooks import DF_S3Hook
from datenfabrik_libraries.helpers import FileAccessS3

@task()
def transform():
    hook = DF_S3Hook(
        aws_conn_id='S3-MINIO',
        extra_args={'Tagging': 'data-type=intermediate'}
    )
    storage = FileAccessS3(hook, bucket="my-bucket")

    # Lesen
    content = storage.read("input/data.csv")

    # Schreiben (CSV)
    with storage.csvwriter("output/result.csv") as writer:
        writer.writerow(["col1", "col2"])

    # Schreiben mit archivierter Kopie
    storage.write(b"...", "output/result.json", archive=True)

Gemeinsamer Code für lokales und S3-Backend

from datenfabrik_libraries.helpers import FileAccess, FileAccessS3

def process(storage):
    data = storage.read_json("config.json")
    with storage.csvwriter("result.csv") as writer:
        writer.writerow(["key", "value"])

# Lokal (z.B. in Tests oder lokaler Ausführung)
process(FileAccess("/tmp/data"))

# S3 (in Produktion / Airflow)
process(FileAccessS3(hook, bucket="my-bucket"))

FileAccessS3 — Archivierung

FileAccessS3.write() und upload_buf() unterstützen den Parameter archive=True. Damit wird zusätzlich zur regulären Datei eine mit Datum versehene Kopie unter archive/<name>_<YYYY-MM-DD>.<ext> gespeichert.

storage.write(result_bytes, "output/report.csv", archive=True)
# Speichert: output/report.csv
# Archiviert: archive/output/report_2026-09-15.csv