Zum Hauptinhalt springen

Generatoren in Datenpipelines

Idee​

from pathlib import Path

def zeilen(pfad: Path):
    with pfad.open(encoding="utf-8") as f:
        for zeile in f:
            yield zeile.rstrip("\n")

def nur_error(zeilen):
    for z in zeilen:
        if "error" in z.lower():
            yield z

def mit_nummer(zeilen):
    for i, z in enumerate(zeilen, start=1):
        yield f"{i}: {z}"

# pipeline
# for out in mit_nummer(nur_error(zeilen(Path("app.log")))):
#     print(out)

Jede Stufe liefert bei Bedarf weiter – super bei großen Dateien.

Rate mal!​

Warum nicht read().splitlines() bei Riesenlogs?

Auflösung

Alles auf einmal im RAM – Generatoren streamen zeilenweise.

Probiere es selbst​

Experiment 1​

zeilen-Generator.

Experiment 2​

Filter-Stufe.

Experiment 3​

Drei Stufen verbinden.

Übungen​

Level 1​

Pipeline skizzieren (3 Kästen).

Level 2​

Filter "ok".

Level 3​

Zähle gefilterte Zeilen mit sum(1 for ...).

Level-3-Lösung
n = sum(1 for z in nur_error(zeilen(Path("app.log"))))
print(n)

Mini-Quiz​

Mini-QuizWas ist ein Vorteil von Generator-Pipelines?

Pipeline sitzt! Mini-Projekt: Log-Analyse.