2025-03-28

DuckDB in Streamlit nutzen

Petrica Leuca

Einführung

Im Artikel Analyzing Railway Traffic in the Netherlands hat Gábor gezeigt, wie man Daten mit SQL analysiert und Remote-Dateien mit DuckDB liest – anhand offener Daten der Anwendung Rijden de Treinen (Fahren die Züge?).

Mit diesen offenen Daten bauen wir in diesem Beitrag eine Anwendung, in der ein Nutzer:

Die Eisenbahnverkehrsdaten verstehen kann;
Den Eisenbahnnetz-Verkehr 2024 mit Plotly visualisieren kann;
Die fünf nächstgelegenen Bahnhöfe zu jedem auf der Karte gewählten Punkt mit der Streamlit-Folium-Komponente visualisieren kann.

Die Anwendung wird mit Streamlit entwickelt, einem Open-Source-Framework, in dem man Datenanwendungen leicht mit Python implementieren kann. Der Anwendungscode ist auf GitHub verfügbar.

Verbindung zu DuckDB in Streamlit

Um die Eisenbahnverkehrsdaten zu laden, schreiben wir eine Funktion, die die zwei Tabellen anlegt, mit denen wir arbeiten werden, services und stations:

def prepare_duckdb(duckdb_conn):

    duckdb_conn.sql("""
        create table if not exists services as
        from 'https://blobs.duckdb.org/nl-railway/services-2024.csv.gz'
    """)

    duckdb_conn.sql("""
        create table if not exists stations as
        from 'https://blobs.duckdb.org/nl-railway/stations-2023-09.csv'
    """)

Aus einer Streamlit-Anwendung gibt es drei Wege, sich mit DuckDB zu verbinden:

In-Memory-Verbindung;
Persistierte lokale Dateiverbindung;
Anhängen einer externen Datenbank.

In-Memory-Verbindung

Eine Verbindung zu einer In-Memory-Datenbank bedeutet, die Daten in den Speicher zu laden, wann immer eine DuckDB-Verbindung hergestellt wird. Abhängig von den Ressourcen des Anwendungsservers und der Datenmenge verlangsamt das Erzeugen einer neuen In-Memory-DuckDB-Verbindung und das Laden der benötigten Daten bei jeder Anfrage die Anwendung. In Streamlit wird dieses Verhalten durch Cachen der Resource adressiert, was so geschehen kann:

Cachen der DuckDB-Verbindung als global geteilte Verbindung in der Anwendung;
Cachen der DuckDB-Verbindung für die Usersession.

@st.cache_resource(ttl=datetime.timedelta(hours=1), max_entries=2)
def get_duckdb_memory(session_id):
    """
    Set a caching resource which will be refreshed
     - either at each hour
     - either at each third call
     - either when the connection is established for a new session_id
    """

    duckdb_conn = duckdb.connect()
    prepare_duckdb(duckdb_conn=duckdb_conn)

    return duckdb_conn

Persistierte lokale Dateiverbindung

Eine DuckDB-Verbindung kann gegen eine persistierte Datenbankdatei auf dem Server der Anwendung hergestellt werden. Die Datenbankdatei kann erzeugt werden:

während des Deployments, bevor die Anwendung startet;
durch einen eigenständigen Prozess, der die Datenbank erzeugt und sie auf den Anwendungsserver kopiert. Abhängig von der Frische-Anforderung der Anwendung könnte man einen Prozess schedulen, der die Daten refresht.

duckdb_conn = duckdb.connect(
    "train_stations_and_services.duckdb",
    read_only=True
)

Anhängen einer externen Datenbank

Ein weiterer Weg, sich mit DuckDB zu verbinden, ist eine Read-only-Verbindung zu einer Instanz über HTTPS oder S3-kompatible APIs herzustellen, z. B. DUCKDB_EXTERNAL_LOCATION = "https://blobs.duckdb.org/nl-railway/train_stations_and_services.duckdb":

duckdb_conn = duckdb.connect()
duckdb_conn.execute(f"attach '{DUCKDB_EXTERNAL_LOCATION}' as ext_db")
duckdb_conn.execute("use ext_db")

Mehr Details zum Anhängen einer externen Datenbank finden Sie in der DuckDB-Dokumentation.

Hinweise

DuckDB funktioniert am besten, wenn Sie 1–4 GB Speicher pro Thread zugewiesen haben; konfigurieren Sie daher die passenden Ressourcen beim Deployment der Streamlit-Anwendung.
Die Empfehlung ist, für jede Datenbankinteraktion eine neue Verbindung zu nutzen oder mindestens eine auf Usersession-Ebene – und nicht eine globale Verbindung auf Anwendungsebene. Eine globale Verbindung kann zu unerwartetem Verhalten führen, abhängig von der Zahl der Nutzer, die die Anwendung gleichzeitig nutzen.
Bei einer persistierten lokalen Datenbank beachten Sie, dass bei einer etablierten Datenbankverbindung kein anderer Prozess darauf schreiben kann. Ein Workaround ist ein Prozess, der die Datenbankdatei auf dem Server erzeugt und ersetzt.
Bei einer angehängten Datenbank beachten Sie, dass die Anwendung je nach gewählter Storage-Lösung einen Performance-Effekt haben kann.

Ab hier arbeiten wir mit der externen Datenbank.

Niederländische Eisenbahndaten analysieren

Um die Daten in Streamlit zu analysieren, nutzen wir die Python-Relational-API, indem wir Queries verketten. Wir beginnen mit der Definition einer Relation, die die benötigten Daten aus stations selektiert:

stations_selection = duckdb_conn.sql("""
    select name_long as station_name, geo_lat, geo_lng, code
    from stations
""").set_alias("stations_selection")

Dann definieren wir eine Relation für die services-Selektion:

services_selection = (
    duckdb_conn.sql("from services")
    .aggregate("""
        station_code: "Stop:Station code",
        service_date: "Service:Date",
        service_date_format: strftime(service_date, '%d-%b (%A)'),
        num_services: count(*)
    """)
    .set_alias("services")
)

Dann joinen wir die zwei Relationen, um den Stationsnamen, die Geolokation und weitere Details zu selektieren, die wir weiter nutzen, und fassen den obigen Code in einer Funktion zusammen.

def get_stations_services_query(duckdb_conn):

    # create a relation for the station selection
    stations_selection = ...

    # create a relation for the services selection
    services_selection = ...

    # return the query and the duckdb_conn
    return (
        stations_selection
        .join(
            services_selection,
            "services.station_code = stations_selection.code"
        )
        .select("""
            service_date,
            service_date_format,
            station_name,
            geo_lat,
            geo_lng,
            num_services
        """)
    ), duckdb_conn

Wenn wir get_stations_services_query(get_duckdb_conn()) ausführen, werden noch keine Daten aus der Datenbank geholt. Das liegt daran, dass die Query-Auswertung lazy ist: Die Query wird erst gegen die Datenbank ausgeführt, wenn ein Ausführungsbefehl auftritt.

Die Query wird ausgeführt, wenn eine der folgenden Methoden auftritt:

stations_query.df(), um in ein Pandas-Dataframe zu extrahieren;
stations_query.fetchall(), um in eine Liste zu extrahieren;
stations_query.write_to(), um die Daten in eine Datei zu exportieren;
jede andere Berechnungsmethode wie .sum, .row_number usw. Die Relational-Methoden finden Sie in der DuckDB-Dokumentation.

Die 5 meistfrequentierten Bahnhöfe pro Monat

Um die 5 meistfrequentierten Bahnhöfe pro Monat zu extrahieren, starten wir von der obigen Query und fügen eine row_number-Berechnung mit einem finalen Filter hinzu, dass das Ranking kleiner oder gleich 5 ist:

stations_query, _ = get_stations_services_query(get_duckdb_conn())

top_5_query = (
    stations_query.aggregate("""
            station_name,
            service_month: monthname(service_date),
            service_month_id: month(service_date),
            num_services: sum(num_services)
        """)
    .select("""
            station_name,
            service_month,
            service_month_id,
            num_services,
            rn: row_number()
                over (
                    partition by service_month
                    order by num_services desc
                )
        """)
    .filter("rn <= 5")
    .order("service_month_id, station_name")
)

Streamlit bietet eine Reihe von Charts wie Linie und Balken, die aber leider nicht viele Personalisierungsmöglichkeiten bieten. Deshalb wurde das obige Chart mit Plotly erzeugt, einer Open-Source-Grafikbibliothek, die mit Streamlit durch den Aufruf st.plotly_chart integriert ist.

Analyse der Eisenbahnnetz-Auslastung

Die Daten über die Zugfahrten durch die Bahnhöfe der Niederlande auf Tagesebene sind ein perfekter Input, um die Auslastung des Eisenbahnnetzes über das Land und über das Jahr zu analysieren.

Eisenbahnnetz-Auslastung über das Land

Um die Netzauslastung über das Land zu analysieren, nutzen wir das Chart density_map von Plotly, das eine Heatmap auf einem Kartenchart erzeugt. Weil Heatmaps am besten mit voraggregierten Daten funktionieren, aggregieren wir zuerst die Zahl der Zugservices und die Geolokation der Station:

stations_df = stations_query.aggregate(
    "geo_lat, geo_lng, num_services: sum(num_services)"
).df()

Interessant zu beobachten: Der Nordosten des Landes ist kaum durch Eisenbahn abgedeckt, und während Groningen und Den Haag über das Jahr genutzt zu werden scheinen, war keiner ihrer Bahnhöfe unter den 5 meistfrequentierten Bahnhöfen pro Monat.

Die Heatmap animieren

Aus dem obigen statischen Chart können wir leicht eine Animation mit Plotly erzeugen, indem wir den Parameter Animation Frame liefern. In unserem Fall erzeugen wir eine Animation der Eisenbahnnetz-Auslastung auf Service-Tagesebene im Juli 2024:

Eisenbahnnetz-Auslastung über das Jahr

Aus Jahressicht sind wir neugierig, wie beschäftigt das Netz in jedem Monat und an jedem Wochentag ist – das lässt sich besser über eine Heatmap darstellen. Wir bereiten die Eingabe für das Chart vor, indem wir die Zahl der Services nach dem Monatsnamen des Servicedatums pivoten und nach dem ISO-Wochentagsnamen gruppieren. Derzeit gibt es keine Pivot-Methode in der Python-Relational-API, wir können aber die von der Relational-API erzeugte SQL-Query als Quelle im Pivot-Roh-SQL nutzen:

@st.cache_data(ttl=3600, max_entries=100)

def get_stations_services_data(_duckdb_conn):
    query = _duckdb_conn.sql("from services").aggregate("""
            service_day: dayname("Service:Date"),
            service_day_isodow: isodow("Service:Date"),
            service_month: monthname("Service:Date"),
            num_services: count(distinct "Service:RDT-ID")
        """)

    return (
        _duckdb_conn.sql(f"""
            pivot ({query.sql_query()})
            on service_month
            using sum(num_services)
            group by service_day, service_day_isodow
            order by service_day_isodow
        """)
        .select(
            "January",
            "February",
            "March",
            "April",
            "May",
            "June",
            "July",
            "August",
            "September",
            "October",
            "November",
            "December",
            "service_day",
        )
        .df()
    )

Wir haben uns entschieden, eine Funktion zu erzeugen, um die cache_data-Funktionalität in Streamlit hervorzuheben. Während die obige Query auf einem MacBook Pro mit 12 GB Speicher in 300 ms ausgeführt wird, wollen wir die Ergebnisse cachen, weil sich das Query-Ergebnis nicht oft ändert. cache_data wird in Streamlit genutzt, um die Ergebnisse einer Funktion zu cachen, und hat wie cache_resource die Attribute Time to Live und Max Entries. Die zwei Caching-Mechanismen verlangen, dass die an die Funktion übergebenen Eingabeparameter hashable sind. Da das DuckDB-Verbindungsobjekt nicht hashable ist, können wir Streamlit sagen, es zu ignorieren, indem wir das Präfix _ hinzufügen.

Wichtig: Die Reihenfolge von Spalten und Werten spielt in einem Heatmap-Chart eine Rolle, und wir müssen den Index der gepivoteten Daten vor dem Plotten zurücksetzen.

Der Code zur Implementierung interaktiver Karten mit Plotly in Streamlit ist auf GitHub verfügbar.

Die 5 nächstgelegenen Stationen mit Folium finden

Die 5 nächstgelegenen Bahnhöfe zur Nutzerauswahl

Ein weiterer Use Case für die Eisenbahnnetz-Daten ist, die nächstgelegenen Bahnhöfe zu einem vom Nutzer auf der Karte gewählten Ort zu finden. Streamlit und Plotly haben die Funktionalität, Klick-Events auf dem Chart zu registrieren, funktionieren aber nur, wenn der Nutzer Punkte wählt, die Teil der Chart-Daten sind. Deshalb nutzen wir streamlit-folium, eine Streamlit-Komponente um Folium, die Klicks auf jedem Punkt der Karte registrieren kann.

Um mit Geodaten in DuckDB zu arbeiten, müssen wir die Spatial-Erweiterung installieren und laden:

duckdb_conn.sql("install spatial;")
duckdb_conn.sql("load spatial;")

Dann definieren wir eine Funktion, die die Query holt, die die fünf nächstgelegenen Bahnhöfe zu einem Ort aus Latitude und Longitude zurückgibt.

def get_closest_stations_query(duckdb_conn, lat, lng):

    stations_selection = duckdb_conn.sql("""
        select name_long as station_name, geo_lat, geo_lng, code
        from stations st
        where exists (
                select count(*)
                from services sv
                where st.code = sv."Stop:Station code"
                having count(*) > 100
            )
    """)

    return (
        stations_selection.project(f"""
            code as station_code,
            station_name,
            geo_lat,
            geo_lng,
            station_geo_point: st_point(geo_lng, geo_lat),
            clicked_geo_point: st_point({lng}, {lat}),
            distance_in_m: st_distance_sphere(
                            st_point(geo_lng, geo_lat),
                            clicked_geo_point
                        ),
            distance_in_km: round(distance_in_m/1000,2)
        """)
        .order("distance_in_km")
        .limit(5)
    )

In der obigen Query nutzen wir st_point, um ein Point-Typ-Feld zu erzeugen, und st_distance_sphere, um die Distanz in Metern zwischen zwei Punkten zu bekommen.

Wichtig: Während wir uns auf einen Geopunkt üblicherweise als [latitude, longitude] beziehen, erzeugen wir den Punkt beim Arbeiten mit Spatial-Erweiterungen üblicherweise als [longitude, latitude]. Das gilt auch für PostGIS.

Wenn ein Nutzer auf die Karte klickt, speichern wir die Klicks im Session State und führen Streamlit erneut aus, um die Karte mit der neuen Nutzerauswahl anzuzeigen. Bei einem Rerun führt Streamlit die gesamte Anwendung für die aktuelle Session erneut aus; das Verständnis von Session State und Caching-Mechanismen ist deshalb essenziell beim Bauen von Streamlit-Anwendungen.

Der Code zur Implementierung interaktiver Karten mit Folium in Streamlit ist auf GitHub verfügbar.

Fazit

In diesem Beitrag haben wir gezeigt, wie DuckDB in Streamlit genutzt werden kann, wie die Python-Relational-API unseren Code durch das Verketten von Queries sauberer machen kann und wie man interaktive Karten mit Plotly und Folium implementiert, während man den Streamlit-Caching-Mechanismus nutzt.