如何使用Python、Dash和RisingWave可視化實時數據
譯文譯者 | 李睿
審校 | 重樓
實時數據對于企業快速做出決策非常重要。企業高管可以直觀地看到這些數據幫助他們更快地做出決策，可以使用各種數據應用程序或儀表板創建數據的可視化表示。Dash是一個開源Python庫，它提供了廣泛的內置組件，用于創建交互式圖表、圖形、表格和其他用戶界面（UI）元素。RisingWave是一個基于SQL的流數據庫，用于實時數據處理。本文將對如何使用Python、Dash開源庫和RisingWave對實時數據實現可視化進行了介紹。
如何實時可視化數據
實時數據是指立即生成和處理的數據，因為它是從不同的數據源收集的。源可以是典型的數據庫（例如Postgres或MySQL），也可以是消息代理（例如Kafka）。實時數據的可視化包括幾個步驟：首先攝取，然后處理，最后在儀表板中顯示這些數據。
在訂單交付數據的情況下，實時可視化這些數據可以為餐廳或配送服務的績效提供有價值的見解。例如，可以使用實時數據來監控訂單交付所需的時間，識別交付過程中的瓶頸，并跟蹤隨著時間變化的訂單量。在處理不斷變化的數據時，很難跟蹤正在發生的一切并識別模式或趨勢。使用Dash和RisingWave等免費工具，可以創建交互式可視化，使用戶能夠探索和分析這些不斷變化的數據。
說到處理數據，人們可能首先想到的編程語言是Python，因為Python有一系列庫。Dash是其中之一，它允許用戶僅使用Python代碼創建具有豐富和可定制用戶界面的數據應用程序。Dash是在Flask、Plotly.js和React.js之上構建的，這些都是流行的Web開發工具，所以用戶不需要知道HTML、CSS或其他JavaScript框架。
RisingWave可以使用來自各種來源的數據流，創建針對復雜查詢進行優化的物化視圖，并使用SQL查詢實時數據。由于RisingWave與PostgreSQL是有線兼容的，可以使用Psycopg2 (Python中的PostgreSQL客戶端庫)驅動程序連接到RisingWave，并進行查詢操作。
可視化訂單交付數據演示
在這個演示教程中，將利用下面的GitHub存儲庫和RisingWave演示，假設所有必要的內容都是使用Docker Compose設置的。可以在官方網站上查看運行RisingWave的其他方法。在這里有一個名為delivery_orders的Kafka主題，其中包含在食品配送網站上放置的每個訂單的事件。每個事件都包含有關訂單的信息，例如訂單ID、餐廳ID和交付狀態。工作負載生成器(稱為Datagen的Python腳本)模擬隨機模擬數據的連續生成，并將其流式傳輸到Kafka主題中。實際上，這些模擬數據可以被來自Web應用程序或后端服務的數據所取代。
要完成這一教程，首先完成以下操作：
- 確保環境中安裝了Docker和Docker Compose。
- 確保環境中安裝了PostgreSQL交互終端psql。詳細說明請參見下載PostgreSQL。
- 為OS. pip命令下載并自動安裝Python 3。
這一教程在Windows操作系統、Docker桌面和安裝的Python 3.10.11版本上進行測試演示。
步驟1：設置RisingWave的演示集群
首先，將RisingWave示例存儲庫克隆到本地環境。
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，也可以參考網站上的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還應該安裝Psycopg2來與RisingWave流數據庫交互：
pip install psycopg2-binary步驟3：創建數據源
要使用RisingWave獲取實時數據，首先需要設置一個數據源。在演示項目中，Kafka應該被定義為數據源。將創建一個名為create-a-source.py的新文件，與連接到RisingWave的Python腳本的integration_tests/delivery目錄相同，并創建一個表來消費和持久化delivery_orders Kafka主題?？梢院唵蔚貙⒁韵麓a復制并粘貼到新文件中。
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.在創建文件之后，運行python create-a-source.py，它將在RisingWave中創建源表。
步驟4：創建物化視圖
接下來，創建一個新的物化視圖，類似于創建表的方式。在此創建一個名為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分鐘內創建的訂單總數，并將結果緩存到物化視圖中。如果發生任何數據更改或新的Kafka主題，RisingWave會自動增加和更新物化視圖的結果。一旦設置了數據源，物化視圖，就可以開始攝取數據，并使用Dash將這些數據可視化。
步驟5：構建Dash應用程序
現在構建Dash應用程序來查詢和可視化RisingWave中的物化視圖內容。可以在20分鐘內跟隨教程Dash來了解Dash的基本構建塊。這一示例的應用程序代碼以表格和圖形兩種格式顯示餐廳訂單數據。請參閱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物化視圖檢索數據，并使用Dash_table.DataTable將其顯示在Dash表中，使用dcc.Graph將數據顯示在條形圖中。該表和條形圖的列與物化視圖的列（'window_start'、'total_order'和'restaurant_id'）相對應，行與物化中的數據相對應。
步驟6：查看結果
可以通過運行上面的dash-example.py腳本并導航到http://localhost:8050/在網絡瀏覽器中（用戶會在終端收到一條消息，告訴轉到這個鏈接）。
總結
總的來說，Dash是一個強大的工具，用于創建需要復雜用戶界面和可視化功能的數據分析視圖，所有這些都使用簡單和優雅的Python編程語言。當將它與RisingWave流數據庫一起使用時，可以深入了解實時數據，這可以幫助人們做出更明智的決策，并采取行動優化性能。
原文標題：Visualize Real-Time Data With Python, Dash, and RisingWave，作者：Bobur Umurzokov
