譯者 | 李睿

審校 | 重樓

實時數(shù)據(jù)對于企業(yè)快速做出決策非常重要。企業(yè)高管可以直觀地看到這些數(shù)據(jù)幫助他們更快地做出決策，可以使用各種數(shù)據(jù)應(yīng)用程序或儀表板創(chuàng)建數(shù)據(jù)的可視化表示。Dash是一個開源Python庫，它提供了廣泛的內(nèi)置組件，用于創(chuàng)建交互式圖表、圖形、表格和其他用戶界面（UI）元素。RisingWave是一個基于SQL的流數(shù)據(jù)庫，用于實時數(shù)據(jù)處理。本文將對如何使用Python、Dash開源庫和RisingWave對實時數(shù)據(jù)實現(xiàn)可視化進行了介紹。

如何實時可視化數(shù)據(jù)

實時數(shù)據(jù)是指立即生成和處理的數(shù)據(jù)，因為它是從不同的數(shù)據(jù)源收集的。源可以是典型的數(shù)據(jù)庫（例如Postgres或MySQL），也可以是消息代理（例如Kafka）。實時數(shù)據(jù)的可視化包括幾個步驟：首先攝取，然后處理，最后在儀表板中顯示這些數(shù)據(jù)。

在訂單交付數(shù)據(jù)的情況下，實時可視化這些數(shù)據(jù)可以為餐廳或配送服務(wù)的績效提供有價值的見解。例如，可以使用實時數(shù)據(jù)來監(jiān)控訂單交付所需的時間，識別交付過程中的瓶頸，并跟蹤隨著時間變化的訂單量。在處理不斷變化的數(shù)據(jù)時，很難跟蹤正在發(fā)生的一切并識別模式或趨勢。使用Dash和RisingWave等免費工具，可以創(chuàng)建交互式可視化，使用戶能夠探索和分析這些不斷變化的數(shù)據(jù)。

說到處理數(shù)據(jù)，人們可能首先想到的編程語言是Python，因為Python有一系列庫。Dash是其中之一，它允許用戶僅使用Python代碼創(chuàng)建具有豐富和可定制用戶界面的數(shù)據(jù)應(yīng)用程序。Dash是在Flask、Plotly.js和React.js之上構(gòu)建的，這些都是流行的Web開發(fā)工具，所以用戶不需要知道HTML、CSS或其他JavaScript框架。

RisingWave可以使用來自各種來源的數(shù)據(jù)流，創(chuàng)建針對復(fù)雜查詢進行優(yōu)化的物化視圖，并使用SQL查詢實時數(shù)據(jù)。由于RisingWave與PostgreSQL是有線兼容的，可以使用Psycopg2 (Python中的PostgreSQL客戶端庫)驅(qū)動程序連接到RisingWave，并進行查詢操作。

可視化訂單交付數(shù)據(jù)演示

在這個演示教程中，將利用下面的GitHub存儲庫和RisingWave演示，假設(shè)所有必要的內(nèi)容都是使用Docker Compose設(shè)置的。可以在官方網(wǎng)站上查看運行RisingWave的其他方法。在這里有一個名為delivery_orders的Kafka主題，其中包含在食品配送網(wǎng)站上放置的每個訂單的事件。每個事件都包含有關(guān)訂單的信息，例如訂單ID、餐廳ID和交付狀態(tài)。工作負載生成器(稱為Datagen的Python腳本)模擬隨機模擬數(shù)據(jù)的連續(xù)生成，并將其流式傳輸?shù)終afka主題中。實際上，這些模擬數(shù)據(jù)可以被來自Web應(yīng)用程序或后端服務(wù)的數(shù)據(jù)所取代。

要完成這一教程，首先完成以下操作：

• 確保環(huán)境中安裝了Docker和Docker Compose。
• 確保環(huán)境中安裝了PostgreSQL交互終端psql。詳細說明請參見下載PostgreSQL。
• 為OS. pip命令下載并自動安裝Python 3。

這一教程在Windows操作系統(tǒng)、Docker桌面和安裝的Python 3.10.11版本上進行測試演示。

步驟1：設(shè)置RisingWave的演示集群

首先，將RisingWave示例存儲庫克隆到本地環(huán)境。

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

然后，進入integration_tests/delivery目錄，并從docker compose文件啟動演示集群。

cd risingwave/integration_tests/delivery
docker compose up -d

確保所有容器都已啟動并運行。

步驟2：安裝Dash和Psycopg2庫

要安裝Dash，也可以參考網(wǎng)站上的Dash安裝指南?；旧?，需要通過運行以下pip install命令來安裝兩個庫(Dash本身和Pandas)：

# 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

還應(yīng)該安裝Psycopg2來與RisingWave流數(shù)據(jù)庫交互：

pip install psycopg2-binary

步驟3：創(chuàng)建數(shù)據(jù)源

要使用RisingWave獲取實時數(shù)據(jù)，首先需要設(shè)置一個數(shù)據(jù)源。在演示項目中，Kafka應(yīng)該被定義為數(shù)據(jù)源。將創(chuàng)建一個名為create-a-source.py的新文件，與連接到RisingWave的Python腳本的integration_tests/delivery目錄相同，并創(chuàng)建一個表來消費和持久化delivery_orders Kafka主題。可以簡單地將以下代碼復(fù)制并粘貼到新文件中。

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.

在創(chuàng)建文件之后，運行python create-a-source.py，它將在RisingWave中創(chuàng)建源表。

步驟4：創(chuàng)建物化視圖

接下來，創(chuàng)建一個新的物化視圖，類似于創(chuàng)建表的方式。在此創(chuàng)建一個名為create-a-materialized-view.py的新文件，并使用Psycopg2庫運行SQL查詢。也可以將上面的最后兩個步驟合并到一個Python腳本文件中。

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

在上面的示例中，SQL查詢實時計算特定餐廳在過去15分鐘內(nèi)創(chuàng)建的訂單總數(shù)，并將結(jié)果緩存到物化視圖中。如果發(fā)生任何數(shù)據(jù)更改或新的Kafka主題，RisingWave會自動增加和更新物化視圖的結(jié)果。一旦設(shè)置了數(shù)據(jù)源，物化視圖，就可以開始攝取數(shù)據(jù)，并使用Dash將這些數(shù)據(jù)可視化。

步驟5：構(gòu)建Dash應(yīng)用程序

現(xiàn)在構(gòu)建Dash應(yīng)用程序來查詢和可視化RisingWave中的物化視圖內(nèi)容?？梢栽?0分鐘內(nèi)跟隨教程Dash來了解Dash的基本構(gòu)建塊。這一示例的應(yīng)用程序代碼以表格和圖形兩種格式顯示餐廳訂單數(shù)據(jù)。請參閱dash-example.py中的以下Python代碼：

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)

這一代碼片段使用Pandas從restaurant_orders_view物化視圖檢索數(shù)據(jù)，并使用Dash_table.DataTable將其顯示在Dash表中，使用dcc.Graph將數(shù)據(jù)顯示在條形圖中。該表和條形圖的列與物化視圖的列（'window_start'、'total_order'和'restaurant_id'）相對應(yīng)，行與物化中的數(shù)據(jù)相對應(yīng)。

步驟6：查看結(jié)果

可以通過運行上面的dash-example.py腳本并導(dǎo)航到http://localhost:8050/在網(wǎng)絡(luò)瀏覽器中（用戶會在終端收到一條消息，告訴轉(zhuǎn)到這個鏈接）。

總結(jié)

總的來說，Dash是一個強大的工具，用于創(chuàng)建需要復(fù)雜用戶界面和可視化功能的數(shù)據(jù)分析視圖，所有這些都使用簡單和優(yōu)雅的Python編程語言。當將它與RisingWave流數(shù)據(jù)庫一起使用時，可以深入了解實時數(shù)據(jù)，這可以幫助人們做出更明智的決策，并采取行動優(yōu)化性能。

原文標題：Visualize Real-Time Data With Python, Dash, and RisingWave，作者：Bobur Umurzokov