Dieser Fall ist am ehesten mit lokalem Training vergleichbar: Eine große Datei in der Cloud, aus der der Trainingsdatensatz eingelesen werden soll. Das kommt oft bei tabellarischen Daten vor, bei denen der gesamte Datensatz in Speicherformaten wie CSV, Parquet etc. zur Verfügung steht.

Im Code unten ist eine Möglichkeit dargestellt, wie Daten flexibel in beliebigen Dateiformaten aus dem Cloud Storage eingelesen werden können. Die BlobHandler-Klasse regelt den Lesevorgang aus der Cloud, eine StreamConverter-Klasse bestimmt, wie das gegebene Dateiformat interpretiert und umgewandelt werden soll. (Hier: Verstehe das Dateiformat als CSV und konvertiere es in einen Pandas DataFrame.)

import io
from abc import ABC, abstractmethod
from typing import Union, Iterable
from google.cloud.storage import Client, Blob


class StreamConverter(ABC):
    @abstractmethod
    def from_stream(self, stream: io.BytesIO) -> Iterable:
        ...


class BlobHandler:
    def __init__(self, bucket: str, io_handler: StreamConverter):
        self.client = Client()
        self.bucket = self.client.get_bucket(bucket)
        self.io_handler = io_handler

    def read_from_blob(self, blob_or_uri: Union[str, Blob]) -> Iterable:
        with io.BytesIO() as stream:
            self.client.download_blob_to_file(blob_or_uri, stream)
            stream.seek(0)
            out = self.io_handler.from_stream(stream)
        return out

import pandas as pd
    
class CSV2PandasConverter(StreamConverter):
    def from_stream(self, stream: io.BytesIO) -> pd.DataFrame:
        return pd.read_csv(stream)

Lesen der Daten in Chunks

Große Datenmengen im Big-Data-Bereich übersteigen schnell die Größe des vorhandenen Arbeitsspeichers, wodurch es unmöglich wird, den gesamten Datensatz auf einmal (auf einer Maschine) zu laden. Das führt zu folgenden Einschränkungen:

1. Die Wahl der Algorithmen wird beschränkt: Viele Modelltypen benötigen den gesamten Datensatz auf einmal, damit sie trainiert werden können. Ist der Datensatz zu groß, ist man auf Modelle angewiesen, die mit Batch- oder Online-Learning trainiert werden können. So wird pro Iteration immer nur ein Teil bzw. nur eine einzelne Stichprobe des Datensatzes benötigt.
2. Das Dateiformat muss dafür ausgelegt sein, portioniert eingelesen zu werden. Voraussetzung hierfür ist:

• Die Datei muss in einem zeilenbasierten Speicherformat vorliegen,
• oder der Datensatz muss bereits in mehreren portionierten Dateien gespeichert sein (viele kleine Dateien).

Grundsätzlich gibt es zwei Methoden, tabellarische Daten zu speichern – spaltenweise (‚columnar‘ – z. B. Parquet) und zeilenweise (z. B. CSV, Avro). Wenn eine oder mehrere Stichproben/Zeilen aus einer Datei gelesen werden sollen, ohne die Datei komplett geladen zu haben, ist das nur mit zeilenbasierten Formaten möglich.

Pythons Iteratoren sind gut dafür geeignet, einen großen Datensatz in Batches portioniert einzulesen. Das nächste Beispiel zeigt, wie der vorhandene Code erweitert werden kann, um eine Datei im Avro-Format einzulesen, sodass sich jeweils nur ein Batch mit definierter Größe im Arbeitsspeicher befindet.

from contextlib import contextmanager
from typing import Iterator
import fastavro

@contextmanager
def stream_from_blob(self, blob_or_uri: Union[str, Blob]) -> Iterator[Iterable]:
    with io.BytesIO() as stream:
        self.client.download_blob_to_file(blob_or_uri, stream)
        stream.seek(0)
        yield self.io_handler.from_stream(stream)
        
BlobHandler.stream_from_blob = stream_from_blob


class Avro2PandasChunckGen(StreamConverter):
    chuncksize: int

    def from_stream(self, stream: io.BytesIO) -> Iterator[pd.DataFrame]:
        record_stream = fastavro.reader(stream)
        while True:
            records = []
            for i, record in zip(range(self.chuncksize), record_stream):
                records.append(record)
            if len(records) == 0:
                break
            yield pd.DataFrame.from_records(records)

Einlesen während des Trainings

Ein grundsätzliches Problem mit großen Datenmengen ist der zeitintensive Lesevorgang. Bei den bisher gezeigten Ansätzen wird davon ausgegangen, dass zuerst der Datensatz (bzw. ein Batch des Datensatzes) eingelesen und im Anschluss das Modell trainiert wird. Ist jedoch das Modell im Stande, mit Batches trainiert zu werden, ist das verschwendete Zeit.

Das Einlesen von Daten ist kaum rechenintensiv und hauptsächlich von der gegebenen Bandbreite abhängig. Daher bietet es sich an, die nächsten Datenzeilen zu laden, während das Modell – auf Basis schon geladener Daten – trainiert wird.

Im Folgenden ist die grundsätzliche Idee skizziert: Ein Producer Thread liest die Daten zeilenweise ein, während ein Consumer die bereits gelesenen Daten zur Verfügung stellt.

from queue import Queue
from threading import Thread

class PrefetchAvro2PandasChunckGen(StreamConverter):
    stack: Queue = Queue()

    def _produce(self, stream):
        for record in fastavro.reader(stream):
            self.stack.put(record)
        self.stack.put(None)

    def _consume(self):
        while record := self.stack.get() is not None:
            yield record

    def from_stream(self, stream: io.BytesIO) -> Iterator:

        producer_thread = Thread(target=lambda: self._produce(stream))
        producer_thread.start()

        for record in self._consume():
            yield record

        producer_thread.join()