Visualiseer Real-Tijd Data met Python, Dash en RisingWave

Realtijdatalagebruikelijkvoorbedrijvenomsnellebeslissingente nemen. Het visueel bekijken van deze gegevens kan beslissingen nog sneller helpen nemen. We kunnen visuele weergaven van gegevens maken met verschillende gegevensapps of dashboards. Dash is een open source Python bibliotheek die een breed scala aan ingebouwde componenten biedt voor het maken van interactieve grafieken, diagrammen, tabellen en andere UI-elementen. RisingWave is een SQL-gebaseerde streaming database voor realtime gegevensverwerking. In dit artikel wordt uitgelegd hoe u Python, Dash en RisingWave kunt gebruiken om visualisaties van realtime gegevens te maken.

Hoe Gegevens Realtijd Visueel Weer Te Geven

We weten dat realtime gegevens gegevens zijn die onmiddellijk worden gegenereerd en verwerkt, zoals ze worden verzameld uit verschillende gegevensbronnen. Bronnen kunnen typische databases zijn zoals Postgres of MySQL, en berichtenbrokers zoals Kafka. Een realtime gegevensvisualisatie bestaat uit een paar stappen: eerst opnemen, dan verwerken, en tenslotte deze gegevens tonen in een dashboard.

In het geval van bestel- en leveringsgegevens, kan het visualiseren van deze gegevens in realtime waardevolle inzichten bieden in de prestaties van een restaurant of leveringsdienst. Zo kunnen we bijvoorbeeld realtimegegevens gebruiken om te monitoren hoe lang het duurt voordat bestellingen worden geleverd, knelpunten in het leveringsproces te identificeren en veranderingen in de bestelvolume over tijd te volgen. Bij het omgaan met voortdurend veranderende gegevens kan het moeilijk zijn om bij te houden wat er allemaal gebeurt en patronen of trends te identificeren. Met gratis hulpmiddelen zoals Dash en RisingWave kunnen we interactieve visualisaties maken die ons in staat stellen om deze voortdurend veranderende gegevens te verkennen en te analyseren.

Wanneer het gaat om het werken met gegevens, is de eerste programmeertaal die je misschien in gedachten haalt Python, omdat het een reeks bibliotheken heeft. Dash is er een die ons in staat stelt om een gegevensapp met een rijke en aanpasbare gebruikersinterface te maken met behulp van alleen Python-code. Dash is gebouwd op basis van Flask, Plotly.js en React.js, die populaire webontwikkelgereedschappen zijn, zodat je geen HTML, CSS of andere JavaScript-frameworks hoeft te kennen.

Met RisingWave kunnen we gegevensstromen consumeren van verschillende bronnen, kunnen we materialiseerbare weergaven maken die zijn geoptimaliseerd voor complexe query’s en realtimegegevens kunnen opvragen met SQL. Omdat RisingWave wire-compatibel is met PostgreSQL, kunnen we de psycopg2 (PostgreSQL-clientbibliotheek in Python) driver gebruiken om verbinding te maken met RisingWave en query-operaties uit te voeren. Zie in de volgende sectie.

Visualiseren van Bestelling Bezorgingsgegevens Demo

In de demo-tutorial gaan we gebruik maken van het volgende GitHub-repository met RisingWave-demos waar we ervan uitgaan dat alle noodzakelijke zaken zijn ingesteld met Docker Compose. U kunt andere manieren om RisingWave uit te voeren op de officiële website bekijken. We hebben een Kafka-onderwerp genaamd delivery_orders dat gebeurtenissen bevat voor elke bestelling die wordt geplaatst op een food delivery website. Elke gebeurtenis bevat informatie over de bestelling, zoals de bestelling ID, restaurant ID, en bezorgingsstatus. De workload generator (Python-script genaamd Datagen) simuleert het genereren van continue willekeurige mock-gegevens en stuurt deze naar Kafka-onderwerpen. In werkelijkheid kan deze mock-gegevens worden vervangen door gegevens die afkomstig zijn van uw webapp of achterkant service.

Voordat u begint

Om deze tutorial te kunnen voltooien, heeft u het volgende nodig:

• Zorg ervoor dat u Docker en Docker Compose in uw omgeving hebt geïnstalleerd.
• Zorg ervoor dat de PostgreSQL interactieve terminal, psql, geïnstalleerd is in uw omgeving. Voor gedetailleerde instructies, zie Download PostgreSQL.
• Download en installeer Python 3 voor uw OS. De pip opdracht wordt automatisch geïnstalleerd.

De demo die ik getest heb op Windows OS, Docker Desktop, en Python 3.10.11-versie geïnstalleerd.

Stap 1: Het opzetten van RisingWave’s Demo Cluster

Kloon eerst de RisingWave voorbeeldrepository naar uw lokale omgeving.

git clone <https://github.com/risingwavelabs/risingwave.git>

Vervolgens, de integration_tests/delivery directory en start de demo cluster vanaf de docker compose file.

cd risingwave/integration_tests/delivery
docker compose up -d

Zorg ervoor dat alle containers actief zijn!

Stap 2: Installeer Dash en Psycopg2 Libraries

Om Dash te installeren, kunt u ook de Dash installatie handleiding op de website raadplegen. In principe moeten we twee bibliotheken (Dash zelf en Pandas) installeren door de volgende pip install opdracht uit te voeren:

# This also brings along the Plotly graphing library.
# Plotly is known for its interactive charts
# Plotly Express requires Pandas to be installed too.
pip install dash pandas

We moeten ook psycopg2 installeren om te kunnen communiceren met de RisingWave streaming database:

pip install psycopg2-binary

Stap 3: Maak een Gegevensbron

Om realtime gegevens met RisingWave te verwerken, moet u eerst een gegevensbron instellen. In het demo project moet Kafka gedefinieerd worden als de gegevensbron. We gaan een nieuw bestand aanmaken genaamd create-a-source.py, in dezelfde integration_tests/delivery directory met Python script waar we verbinding maken met RisingWave, en een tabel aanmaken om de delivery_orders Kafka onderwerpen te consumeren en te behouden. U kunt gewoon de onderstaande code in het nieuwe bestand kopiëren en plakken.

import psycopg2

conn = psycopg2.connect(database="dev", user="root", password="", host="localhost", port="4566") # Connect to RisingWave.

conn.autocommit = True # Set queries to be automatically committed.

with conn.cursor() as cur:
    cur.execute("""
CREATE TABLE delivery_orders_source (
    order_id BIGINT,
    restaurant_id BIGINT,
    order_state VARCHAR,
    order_timestamp TIMESTAMP
) WITH (
    connector = 'kafka',
    topic = 'delivery_orders',
    properties.bootstrap.server = 'message_queue:29092',
    scan.startup.mode = 'earliest'
) ROW FORMAT JSON;""") # Execute the query.

conn.close() # Close the connection.

Nadat u het bestand hebt aangemaakt, voert u python create-a-source.py uit en zal het de bron tabel in RisingWave aanmaken.

Stap 4: Maak een Materialized View

Vervolgens creëren we een nieuwe materialized view op een manier die vergelijkbaar is met hoe we een tabel hebben gemaakt. We maken een nieuw bestand met de naam create-a-materialized-view.py en voeren een SQL-query uit met behulp van de psycopg2 bibliotheek. Het is ook mogelijk om de bovenstaande laatste twee stappen samen te voegen in één Python-scriptbestand.

import psycopg2

conn = psycopg2.connect(database="dev", user="root", password="", host="localhost", port="4566")
conn.autocommit = True

with conn.cursor() as cur:
    cur.execute("""CREATE MATERIALIZED VIEW restaurant_orders_view AS
SELECT
    window_start,
    restaurant_id,
    COUNT(*) AS total_order
FROM
    HOP(delivery_orders_source, order_timestamp, INTERVAL '1' MINUTE, INTERVAL '15' MINUTE)
WHERE
        order_state = 'CREATED'
GROUP BY
    restaurant_id,
    window_start;""")

conn.close()

Bovenstaand SQL-query berekent het aantal totaal bestellingen dat is aangemaakt door een specifiek restaurant binnen de afgelopen 15 minuten in realtime en cachet het resultaat in de materialized view. Als er gegevenswijzigingen plaatsvinden of nieuwe Kafka-onderwerpen arriveren, verhoogt en werkt RisingWave automatisch het resultaat van de materialized view bij. Zodra je de gegevensbron, materialized view hebt ingesteld, kun je beginnen met het opnemen van gegevens en deze visualiseren met behulp van Dash.

Stap 5: Het bouwen van een Dash-app

Nu bouwen we onze Dash-app om de inhoud van de materialized view in RisingWave te queryen en te visualiseren. Je kunt de tutorial Dash in 20 minuten volgen om de basisbouwstenen van Dash te begrijpen. Ons voorbeeldapplicatiecode toont restaurantbestellingsgegevens in zowel tabel- als grafiekvorm. Zie onderstaande Python-code in dash-example.py:

import psycopg2
import pandas as pd
import dash
from dash import dash_table
from dash import dcc
import dash_html_components as html
import plotly.express as px

# Connect to the PostgreSQL database
conn = psycopg2.connect(database="dev", user="root", password="", host="localhost", port="4566")

# Retrieve data from the materialized view using pandas
df = pd.read_sql_query("SELECT window_start, restaurant_id, total_order FROM restaurant_orders_view;", conn)

# Create a Dash application
app = dash.Dash(__name__)

# Define layout
app.layout = html.Div(children=[
    html.H1("Restaurant Orders Table"),
    dash_table.DataTable(id="restaurant_orders_table", columns=[{"name": i, "id": i} for i in df.columns], data=df.to_dict("records"), page_size=10),
    html.H1("Restaurant Orders Graph"),
    dcc.Graph(id="restaurant_orders_graph", figure=px.bar(df, x="window_start", y="total_order", color="restaurant_id", barmode="group"))
])

# Run the application
if __name__ == '__main__':
    app.run_server(debug=True)

Dit codefragment haalt gegevens op uit de restaurant_orders_view materiële weergave met behulp van pandas en geeft deze weer in een Dash-tabel met dash_table.DataTable en een staafdiagram met dcc.Graph. De tabel en het staafdiagram hebben kolommen die overeenkomen met de kolommen in de materiële weergave (‘window_start’, ’total_order’, en ‘restaurant_id’) en rijen die overeenkomen met de gegevens in de materiële weergave.

Stap 6: Bekijk de Resultaten

U kunt de toepassing uitvoeren door het bovenstaande dash-example.py script uit te voeren en naar http://localhost:8050/ in uw webbrowser te navigeren (u krijgt een bericht in de terminal dat u naar deze link moet gaan).

Samenvatting

Over het algemeen is Dash een krachtig hulpmiddel voor het maken van gegevensanalytische weergaven die complexe UIs en gegevensvisualisatie mogelijkheden vereisen, allemaal gebruik makend van de eenvoud en elegantie van de Python-programmeertaal. Wanneer we het samen gebruiken met RisingWave streaming database, krijgen we inzicht in realtime gegevens die ons kunnen helpen om beter geïnformeerde beslissingen te nemen en actie te ondernemen om de prestaties te optimaliseren.

Gerelateerde Bronnen

• Realtime gegevensanalyse met Apache Superset, Redpanda en RisingWave
• Hoe live stream metrische gegevens te bewaken