2025-05-16

Machine-Learning-Prototyping mit DuckDB und scikit-learn

Petrica Leuca

Einleitung

Machine-Learning-Prototyping bedeutet oft, Datensätze, Preprocessing-Schritte und Performance-Grenzen gleichzeitig zu jonglieren – das macht den Prozess komplex und zeitaufwendig. scikit-learn ist eine der beliebtesten und leistungsfähigsten Python-Bibliotheken für Machine Learning und bietet eine große Menge an Werkzeugen zum Bauen und Bewerten von Modellen. In diesem Beitrag schauen wir, wie DuckDB scikit-learn im Modellentwicklungszyklus ergänzt, indem wir ein Vorhersagemodell für Pinguinarten auf Beobachtungen im Palmer-Archipel umsetzen.

Die folgende Implementierung läuft in einem marimo-Python-Notebook, das auf GitHub in unserem Examples-Repository verfügbar ist.

Datenvorbereitung

Wir beginnen mit dem Laden des Palmer-Penguins-Datensatzes und nutzen DuckDBs COLUMNS(*)-Ausdruck, um jeden Datensatz herauszufiltern, der NA oder NULL enthält:

duckdb_conn.read_csv(
    "https://blobs.duckdb.org/data/penguins.csv"
).filter(
    "columns(*)::text != 'NA'"
).filter(
    "columns(*) is not null"
).select(
    "*, row_number() over () as observation_id"
).to_table(
    "penguins_data"
)

Auch wenn die NA-Werte aus dem Datensatz entfernt sind, hat DuckDB den Spaltentyp beim read_csv als VARCHAR erkannt. Deshalb ändern wir den Spaltentyp mit der Anweisung ALTER TABLE auf einen numerischen Typ, DECIMAL(5, 2):

duckdb_conn.sql(
    "alter table penguins_data alter bill_length_mm set data type decimal(5, 2)"
)

Tipp Man kann das Schema auch schon beim Import definieren.

Jetzt können wir die Daten plotten und nach artspezifischen Clustern schauen. Mit einem Scatter Plot etwa erkennen wir Cluster bei der Kombination aus Schnabeltiefe und Schnabellänge:

Pinguinbeobachtungen, Schnabeltiefe und -länge

Wir sehen außerdem ein paar deskriptive Spalten im Datensatz, etwa species und island. In einem Machine-Learning-Workflow ist ein häufiger Preprocessing-Schritt, solche Werte in numerische Werte umzuwandeln und ihnen eine eindeutige Kennung zuzuweisen; das heißt Label Encoding. scikit-learn bietet zwar ein LabelEncoder-Utility, der Prozess ähnelt aber stark der Arbeit mit Referenztabellen in einem Data Warehouse. Deshalb definieren wir eine Funktion, die für jede Spalte (jedes Feature) eine Referenztabelle anlegt und dabei die DuckDB-Python-Relational API nutzt:

def process_reference_data(duckdb_conn):
    for feature in ["species", "island"]:
        duckdb_conn.sql(f"drop table if exists {feature}_ref")
        (
            duckdb_conn.table("penguins_data")
            .select(feature)
            .unique(feature)
            .row_number(
                window_spec=f"over (order by {feature})",
                projected_columns=feature
            )
            .select(f"{feature}, #2 - 1 as {feature}_id")
            .to_table(f"{feature}_ref")
        )
        duckdb_conn.table(f"{feature}_ref").show()

Nach der Ausführung dieser Funktion entstehen zwei Tabellen mit den Distinct-Werten der Kategorie und einer eindeutigen Kennung, z. B. species_ref:

┌───────────┬────────────┐
│  species  │ species_id │
│  varchar  │   int64    │
├───────────┼────────────┤
│ Adelie    │          0 │
│ Chinstrap │          1 │
│ Gentoo    │          2 │
└───────────┴────────────┘

Der letzte Schritt der Datenvorbereitung ist eine Selection-Query-Variable, die die Daten aus dem Ausgangsdatensatz und den Referenztabellen selektiert:

selection_query = (
    conn.table("penguins_data")
    .join(conn.table("island_ref"), condition="island")
    .join(conn.table("species_ref"), condition="species")
)

selection_query wird lazy ausgewertet, es werden also zum Definitionszeitpunkt keine Daten abgerufen.

Modelltraining

Unser Ziel ist, die Art eines Pinguins anhand seiner Merkmale (Features) vorherzusagen, etwa Schnabellänge und -tiefe, Flossenlänge, Körpermasse und Insel. Ein solches Modell heißt Klassifikationsmodell, weil es die Kategorie vorhersagt, zu der die Eingabedaten gehören. scikit-learn bietet mehrere Klassifikationsmodelle. Für unseren Datensatz haben wir uns für einen Random-Forest-Classifier auf Basis von Entscheidungsbäumen entschieden:

Wir beginnen damit, die Daten in Trainings- und Testdaten zu splitten, mit dem train_test_split-Utility aus scikit-learn:

def train_split_data(selection_query):
    X_df = selection_query.select("""
        bill_length_mm,
        bill_depth_mm,
        flipper_length_mm,
        body_mass_g,
        island_id,
        observation_id,
        species_id
    """).order("observation_id").df()
    y_df = [
        x[0]
        for x in selection_query.order("observation_id").select("species_id").fetchall()
    ]

    num_test = 0.30
    return train_test_split(X_df, y_df, test_size=num_test)

X_train, X_test, y_train, y_test = train_split_data(selection_query)

Das Splitting der Daten ist ein üblicher Schritt in einem Machine-Learning-Workflow und liefert:

X_train, die Eingabedaten, anhand derer wir eine Art-Kategorie zuweisen wollen;
y_train, die Art für jeden Datensatz in X_train;
X_test, die Eingabedaten, anhand derer wir das Modell testen;
y_test, die Art für jeden Datensatz in X_test.

Dann definieren wir den RandomForestClassifier und fitten X_train und y_train darauf. Das Modell speichern wir in einer pickle-Datei, damit wir es ohne erneutes Training nutzen können:

model = RandomForestClassifier(n_estimators=1, max_depth=2, random_state=5)

model.fit(X_train.drop(["observation_id", "species_id"], axis=1).values, y_train)

pickle.dump(model, open("./model/penguin_model.sav", "wb"))

Jetzt können wir den Accuracy-Score des Modells prüfen:

model.score(
    X_test.drop(["observation_id", "species_id"], axis=1).values, y_test
)

0.98

Wir arbeiten mit pickle wegen der geringen Größe unseres Modells. Andere Persistenzmethoden stehen auf der scikit-learn-Dokumentationsseite.

Inferenz mit DuckDB

Inferenz ist der Prozess, das Modell zu nutzen, um Vorhersagen auf (neuen) Daten zu erhalten:

model = pickle.load(open("./model/penguin_model.sav", "rb"))

model.predict(...)

Mit DuckDB gibt es drei Ansätze, die vorhergesagte Kategorie aus den Daten zu holen. Wir können Pandas nutzen oder eine DuckDB-Python-UDF mit oder ohne Batching. Im Folgenden stellen wir diese Ansätze vor.

Mit Pandas

Die häufigste Methode, Vorhersagen abzurufen, ist, sie in eine neue Spalte eines Pandas-Dataframes zu laden:

predicted_df = selection_query.select("""
    bill_length_mm,
    bill_depth_mm,
    flipper_length_mm,
    body_mass_g,
    island_id,
    observation_id,
    species_id
""").df()

predicted_df["predicted_species_id"] = model.predict(
    predicted_df.drop(["observation_id", "species_id"], axis=1).values
)

Mit DuckDB lässt sich der Dataframe per SQL abfragen:

(
    duckdb_conn.table("predicted_df")
    .select("observation_id", "species_id", "predicted_species_id")
    .filter("species_id != predicted_species_id")
)

ergibt:

┌────────────────┬────────────┬──────────────────────┐
│ observation_id │ species_id │ predicted_species_id │
│     int64      │   int64    │        int64         │
├────────────────┼────────────┼──────────────────────┤
│             13 │          1 │                    2 │
│             15 │          1 │                    2 │
│             39 │          1 │                    2 │
│             44 │          1 │                    2 │
│             68 │          1 │                    2 │
│             70 │          1 │                    2 │
│             76 │          1 │                    2 │
│             90 │          1 │                    2 │
│             94 │          1 │                    2 │
│            104 │          1 │                    2 │
│            106 │          1 │                    2 │
│            110 │          1 │                    2 │
│            124 │          1 │                    2 │
│            126 │          1 │                    2 │
│            243 │          3 │                    2 │
│            296 │          2 │                    1 │
├────────────────┴────────────┴──────────────────────┤
│ 16 rows                                  3 columns │
└────────────────────────────────────────────────────┘

Warnung Gibt es eine Tabelle mit demselben Namen wie der Dataframe, kann register genutzt werden, um dem Dataframe einen anderen Tabellennamen zu geben, z. B. duckdb_conn.register("table_name", predicted_df)

DuckDB-Python-UDF, Zeile für Zeile

DuckDB erlaubt es, User Defined Functions (UDFs) aus Python-Funktionen zu registrieren. Weil die UDF zeilenweise ausgeführt wird, gibt unsere Vorhersagefunktion die vorhergesagte Species-ID für jede Zeile zurück:

def get_prediction_per_row(
    bill_length_mm: Decimal,
    bill_depth_mm: Decimal,
    flipper_length_mm: int,
    body_mass_g: int,
    island_id: int
) -> int:
    model = pickle.load(open("./model/penguin_model.sav", "rb"))
    return int(
        model.predict(
            [
                [
                    bill_length_mm,
                    bill_depth_mm,
                    flipper_length_mm,
                    body_mass_g,
                    island_id,
                ]
            ]
        )[0]
    )

In der obigen Python-Funktion übergeben wir die benötigten Features als Eingabe für das Modell und geben den vorhergesagten Wert zurück (ein integer). Mit diesen Informationen erzeugen wir die DuckDB-Funktion:

duckdb_conn.create_function(
    "predict_species_per_row", get_prediction_per_row, return_type=int
)

Die UDF können wir jetzt in der SQL-SELECT-Klausel nutzen:

selection_query.select("""
    observation_id,
    species_id,
    predict_species_per_row(
        bill_length_mm,
        bill_depth_mm,
        flipper_length_mm,
        body_mass_g,
        island_id
    ) as predicted_species_id
""").filter("species_id != predicted_species_id")

Das liefert dieselben Ergebnisse wie oben.

DuckDB-Python-UDF, Batch-Stil

Die zeilenweise Vorhersage ist nützlich, wenn man die Vorhersage nur für eine begrenzte Zahl von Zeilen braucht, wird aber bei mehr Daten weniger performant. Indem wir die Daten in Arrays aggregieren (die die Reihenfolge erhalten), können wir einen Massenabruf (oder Batch-Stil) der Vorhersage simulieren.

Zuerst erzeugen wir eine Python-Funktion, die Vorhersagen für ein JSON-Objekt holt, das die spaltenweise Darstellung der als Modelleingabe nötigen Features enthält:

def get_prediction_per_batch(input_data: dict[str, list[Decimal | int ]]) -> np.ndarray:
    """
    input_data example:
        {
            "bill_length_mm": [40.5],
            "bill_depth_mm": [41.5],
            "flipper_length_mm: [250],
            "body_mass_g": [3000],
            "island_id": [1]
        }
    """
    model = pickle.load(open("./model/penguin_model.sav", "rb"))

    input_data_parsed = orjson.loads(input_data)

    input_data_converted_to_numpy = np.stack(tuple(input_data_parsed.values()), axis=1)

    return model.predict(input_data_converted_to_numpy)


duckdb_conn.create_function(
    "predict_species_per_batch",
    get_prediction_per_batch,
    return_type=duckdb.typing.DuckDBPyType(list[int]),
)

DuckDB hat zwar die Datentypen MAP und STRUCT, die automatisch in ein Dict konvertieren, sie haben aber eine langsamere Ausführungszeit als json_object (einschließlich der orjson-Deserialisierungszeit).

Mit json_object extrahieren wir die spaltenweise Darstellung der Features im Format 'feature_name': array[feature]:

json_object(
    'bill_length_mm', array_agg(bill_length_mm),
    'bill_depth_mm', array_agg(bill_depth_mm),
    'flipper_length_mm', array_agg(flipper_length_mm),
    'body_mass_g', array_agg(body_mass_g),
    'island_id', array_agg(island_id)
) as input_data,

Dann packen wir in einem STRUCT die Vorhersagen zusammen mit anderen Spalten, die uns interessieren:

struct_pack(
    observation_id := array_agg(observation_id),
    species_id := array_agg(species_id),
    predicted_species_id := predict_species_per_batch(input_data)
) as output_data

Zuletzt unnesten wir die Ergebnisse, um die Listen in Tabellen zu flachen:

.select("""
    unnest(output_data.observation_id) as observation_id,
    unnest(output_data.species_id) as species_id,
    unnest(output_data.predicted_species_id) as predicted_species_id
""")

Die obige Query ist in eine Python-Funktion namens get_selection_query_for_batch verpackt, mit der wir Queries verketten können, z. B. für den Massenabruf falscher Vorhersagen:

get_selection_query_for_batch(selection_query).filter("species_id != predicted_species_id")

Ein Batch-Ansatz lässt sich mit LIMIT und OFFSET umsetzen, um die Daten durchzulaufen:

for i in range(4):
    (
        get_selection_query_for_batch(
            selection_query
            .order("observation_id")
            .limit(100, offset=100*i)
            .select("*")
        )
        .filter("species_id != predicted_species_id")
    ).show()

LIMIT und OFFSET werden zuletzt ausgeführt, sie sollten also vor der Vorhersage-Selektion angewendet werden.

Performance-Überlegungen

Um Performance-Daten auf einem größeren Datensatz zu erhalten, haben wir einen Dummy-Datensatz mit etwa 59 Millionen Datensätzen erzeugt. Auf einem Sample von 10 %, auf einem 16-GB-MacBook Pro, liegt die Batch-Verarbeitung zwischen 3 und 4 Sekunden, während die Pandas-Implementierung unter 1 Sekunde läuft. Der Grund: Die Python-UDF enthält mehrere Konvertierungsschritte, die die Performance belasten:

Parsen der Eingabedaten als JSON;
Konvertieren des JSON in ein NumPy-Array;
Unnesten des Arrays in Zeilen.

Wir haben die Python-UDFs trotzdem gezeigt, weil sie in einer Python-Umgebung ein starkes Werkzeug sind und der Performanceunterschied bei Small Data vernachlässigbar ist.

Fazit

In diesem Beitrag haben wir gezeigt, wie DuckDB scikit-learn im Machine-Learning-Entwicklungszyklus ergänzt, mit Fokus auf Datenvorbereitung und Inferenz. Die Inferenz-Ergebnisse auf den Dummy-Daten sind schlecht; liegt das am Modell oder an den Dummy-Daten? Diese Frage lassen wir offen – als Anstoß, DuckDB in den Phasen Modellbewertung und Modelloptimierung zu erkunden.