This case is most comparable to local training: The training data set is read from a large file in the cloud. This often happens with tabular data in whose case the entire data set is available in storage formats such as CSV, Parquet etc.

The code below shows a way to flexibly read data in any file format from cloud storage. The BlobHandler class manages operations for reading from the cloud, while the StreamConverter class determines how the given file format is interpreted and converted (the file format here is CSV and converted into a 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)

Reading data in chunks

Large data volumes in big-data applications quickly exceed available memory capacity, making it impossible to load entire data sets at once (on one machine). This results in the restrictions mentioned next.

1. Limited choice of algorithms: For the purpose of training, many model types require the entire data set at once. If the data set is too large, one has to rely on models which can be trained via batch or online learning. Only a part or single sample of the data set is thus required per iteration.
2. The file format must be designed to be read in portions. A prerequisite for this:

• The file must have been stored in a row-based format.
• Alternatively, the data set must already be stored in several portioned files (many small files).
  Basically, there are two methods of storing tabular data – columnar (e.g. Parquet) and row-by-row (e.g. CSV, Avro). If one or more samples/rows are to be read from a file without it having been fully loaded, this is only possible with row-based formats.

Python iterators are well suited for reading large data sets portioned in batches. The next example shows how our existent code can be extended to read a file in Avro format, so that there is only one batch of a defined size in the main memory at a time.

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)

Reading during training

A fundamental problem with large amounts of data is the time-consuming reading process. The techniques described so far assume that the data set (or a batch of it) is first read, and the model trained subsequently. However, training with a model in batches wastes time.

Reading of data is not very intensive computationally, and depends mainly on available bandwidth. It therefore makes sense to load the next rows of data while the model is being trained on the basis of already loaded data.

The basic idea is outlined next. A Produce thread reads data row by row, while a Consume thread provides data already read.

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