Der DataLoader

Bis hierhin steckte der ganze Datensatz in einem einzigen Tensor. Das funktioniert bei 64 Zeilen und hört bei 64 000 auf zu funktionieren: Der Vorwärtslauf müsste alle Zwischenergebnisse gleichzeitig im Speicher halten, und der Rückwärtslauf noch einmal.

Die Lösung ist banal und heißt in jedem Framework gleich: Batch. Nimm 32 Zeilen, mach damit einen kompletten Schritt, nimm die nächsten 32. Ein Durchgang durch alle Häppchen ist eine Epoche. Nebenbei fällt ein Vorteil ab, den niemand geplant hat: Viele kleine Schritte auf wechselnden Ausschnitten lernen oft besser als wenige große auf allem - das Rauschen hilft.

Das Ziel

Training ist zur Hälfte Daten-Klempnerei. Die Einsicht dieses Kapitels ist weniger eine über Mathematik als eine über den Beruf: Der Teil, der in Vorträgen vorkommt, sind die Gradienten. Der Teil, der die Woche frisst, ist der hier - Formen, Reihenfolgen, Indizes, der letzte unvollständige Batch.

Drei Dinge muss ein DataLoader können, und alle drei sind Indexarbeit:

Schneiden. Aus n Zeilen werden ceil(n / batch_size) Blöcke. Der letzte darf kleiner sein - wegwerfen wäre die schlechtere Wahl, weil dann bei ungünstiger Größe systematisch dieselben Beispiele fehlen.

Mischen. Vor jeder Epoche neu. Sonst sieht das Netz die Beispiele immer in derselben Reihenfolge, und wenn die Daten sortiert vorliegen - erst alle Nullen, dann alle Einsen -, besteht ein Batch aus lauter derselben Klasse. Das Netz lernt dann abwechselnd „alles ist null“ und „alles ist eins“.

Paare wahren. x und y müssen mit derselben Permutation gemischt werden. Zwei getrennte Mischungen ergeben Beispiele mit fremden Antworten - ein Fehler, der nicht abstürzt, sondern nur dafür sorgt, dass nichts lernt.

Zero Magic in diesem Schritt

NumPy ist erlaubt, torch, jax, tinygrad und autograd sind verboten - und diesmal ausdrücklich auch torch.utils.data. NumPy darf hier mehr als sonst: Das Mischen der Indizes ist reine Datenverwaltung und findet vor dem Graphen statt, nicht in ihm. Nur was mit x und y gerechnet wird, muss durch Tensor-Operationen laufen.

Die Aufgabe

Zwei Dateien. Zuerst meintorch/data.py:

import numpy as np

from .tensor import Tensor


class DataLoader:
    """Liefert (x, y)-Batches als Tensoren. Eine Iteration ist eine Epoche."""

    def __init__(self, x, y, batch_size=32, shuffle=True, seed=None):
        self.x = np.asarray(x, dtype=np.float64)
        self.y = np.asarray(y, dtype=np.float64)
        self.batch_size = batch_size
        self.shuffle = shuffle
        self.rng = np.random.default_rng(seed)

    def __len__(self):
        # TODO(du): Anzahl der Batches. Der letzte darf kleiner sein -
        # er zählt trotzdem mit.
        raise NotImplementedError

    def __iter__(self):
        # TODO(du): Indizes bilden, bei shuffle mischen, in Blöcke der
        # Länge batch_size schneiden und je Block (Tensor(x), Tensor(y))
        # herausgeben. x und y mit denselben Indizes.
        raise NotImplementedError

Dann eine zweite Trainingsfunktion in meintorch/train.py, die über Batches läuft statt über den ganzen Datensatz:

def fit(model, loader, epochs=10, lr=0.05, loss_fn=mse_loss):
    """Wie train(), aber je Epoche ein Durchgang durch alle Batches."""
    opt = SGD(model.parameters(), lr=lr)
    history = []
    for _ in range(epochs):
        # TODO(du): über den Loader iterieren, je Batch dieselben fünf
        # Zeilen wie in Kapitel 06, die Loss-Werte einer Epoche mitteln
        # und den Mittelwert anhängen.
        raise NotImplementedError
    return history

Hinweis 1 - die Richtung

__iter__ ist ein Generator: yield statt return, dann funktioniert for xb, yb in loader von selbst.

Der Trick beim Schneiden ist, nicht die Daten zu mischen, sondern die Indizes. Ein Array 0..n-1, einmal gemischt, danach in Scheiben geschnitten - und jede Scheibe ist ein Auswahlindex für x und für y gleichermaßen. Damit ist die Paarbindung geschenkt, statt bewacht werden zu müssen.

Für fit: Der innere Block ist wörtlich der aus Kapitel 06, nur mit xb, yb statt x, y. Der Unterschied ist allein die zusätzliche Schleife drumherum.

Hinweis 2 - konkreter

• __len__: int(np.ceil(len(self.x) / self.batch_size)).
• Indizes: idx = np.arange(len(self.x)), dann self.rng.shuffle(idx), wenn self.shuffle gesetzt ist. shuffle arbeitet in-place.
• Schneiden: for start in range(0, len(idx), self.batch_size), darin teil = idx[start : start + self.batch_size].
• Ausliefern: yield Tensor(self.x[teil]), Tensor(self.y[teil]).
• In fit: summe += float(loss.data) je Batch, am Ende der Epoche history.append(summe / len(loader)).

    # in data.py
    def __len__(self):
        return int(np.ceil(len(self.x) / self.batch_size))

    def __iter__(self):
        idx = np.arange(len(self.x))
        if self.shuffle:
            self.rng.shuffle(idx)
        for start in range(0, len(idx), self.batch_size):
            teil = idx[start : start + self.batch_size]
            yield Tensor(self.x[teil]), Tensor(self.y[teil])


# in train.py
def fit(model, loader, epochs=10, lr=0.05, loss_fn=mse_loss):
    opt = SGD(model.parameters(), lr=lr)
    history = []
    for _ in range(epochs):
        summe = 0.0
        for xb, yb in loader:
            loss = loss_fn(model(xb), yb)
            opt.zero_grad()
            loss.backward()
            opt.step()
            summe += float(loss.data)
        history.append(summe / len(loader))
    return history

Im Goal-Modus

Der autonome Durchlauf muss die Abdeckung beweisen, nicht nur die Formen prüfen:

Lege `meintorch/data.py` an mit einer Klasse
`DataLoader(x, y, batch_size=32, shuffle=True, seed=None)`. `__iter__` mischt
die Indizes (nicht die Daten), schneidet sie in Blöcke und liefert je Block
`(Tensor(x_batch), Tensor(y_batch))`; x und y werden mit derselben Permutation
ausgewählt. `__len__` gibt die Anzahl der Batches zurück, der letzte darf
kleiner sein. Ergänze `meintorch/train.py` um
`fit(model, loader, epochs=10, lr=0.05, loss_fn=mse_loss)`, das je Epoche über
alle Batches läuft und den mittleren Loss je Epoche zurückgibt.
Zwingend: NumPy erlaubt, torch/jax/tinygrad/autograd verboten, auch kein
torch.utils.data.
Schreibe `tests/test_07_dataloader.py`: Die Batches einer Epoche müssen den
Datensatz genau einmal abdecken, der letzte Batch bei 10 Zeilen und
batch_size=3 die Größen [3, 3, 3, 1] ergeben, x und y gepaart bleiben, Mischen
die Reihenfolge ändern und den Inhalt nicht, und `fit` den Loss senken.
Führe `python -m pytest` aus, bis alles grün ist.

Der Meilenstein-Test

tests/test_07_dataloader.py. Der Meilenstein ist der erste Test: Alle Batches zusammen ergeben wieder genau den Datensatz - kein Beispiel doppelt, keines vergessen. Die Testdaten sind absichtlich durchnummeriert, damit jede Zeile an ihrem Wert erkennbar bleibt.

import numpy as np

from meintorch.data import DataLoader
from meintorch.nn import Linear, ReLU, Sequential
from meintorch.train import fit


def daten(n=10):
    # Zeile i ist [2i, 2i+1] mit Antwort i - jede Zeile ist wiedererkennbar
    x = np.arange(n * 2, dtype=float).reshape(n, 2)
    y = np.arange(n, dtype=float).reshape(n, 1)
    return x, y


def test_die_batches_decken_den_datensatz_genau_einmal_ab():
    x, y = daten(10)
    gesehen = np.concatenate(
        [xb.data for xb, _ in DataLoader(x, y, batch_size=3, seed=0)]
    )
    assert gesehen.shape == x.shape
    assert np.allclose(np.sort(gesehen[:, 0]), np.sort(x[:, 0]))


def test_der_letzte_batch_darf_kleiner_sein():
    x, y = daten(10)
    loader = DataLoader(x, y, batch_size=3, shuffle=False)
    assert [xb.shape[0] for xb, _ in loader] == [3, 3, 3, 1]
    assert len(loader) == 4


def test_x_und_y_bleiben_gepaart():
    x, y = daten(10)
    for xb, yb in DataLoader(x, y, batch_size=4, seed=1):
        assert np.allclose(xb.data[:, 0] / 2.0, yb.data[:, 0])


def test_mischen_aendert_die_reihenfolge_nicht_den_inhalt():
    x, y = daten(20)
    gemischt = np.concatenate(
        [xb.data for xb, _ in DataLoader(x, y, batch_size=5, seed=1)]
    )
    geordnet = np.concatenate(
        [xb.data for xb, _ in DataLoader(x, y, batch_size=5, shuffle=False)]
    )
    assert not np.allclose(gemischt, geordnet)
    assert np.allclose(np.sort(gemischt[:, 0]), np.sort(geordnet[:, 0]))


def test_fit_laeuft_ueber_die_batches_und_senkt_den_loss():
    rng = np.random.default_rng(0)
    x = rng.normal(size=(64, 2))
    y = 2.0 * x[:, :1] - x[:, 1:] + 0.5
    netz = Sequential(Linear(2, 8, seed=0), ReLU(), Linear(8, 1, seed=1))
    verlauf = fit(netz, DataLoader(x, y, batch_size=16, seed=0), epochs=30, lr=0.05)
    assert len(verlauf) == 30
    assert verlauf[-1] < verlauf[0] / 5