譯者 | 李睿

審校 | 重樓

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

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

實(shí)時(shí)數(shù)據(jù)是指立即生成和處理的數(shù)據(jù)，因?yàn)樗菑牟煌臄?shù)據(jù)源收集的。源可以是典型的數(shù)據(jù)庫(kù)（例如Postgres或MySQL），也可以是消息代理（例如Kafka）。實(shí)時(shí)數(shù)據(jù)的可視化包括幾個(gè)步驟：首先攝取，然后處理，最后在儀表板中顯示這些數(shù)據(jù)。

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

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

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

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

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

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

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

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

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

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

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

然后，進(jìn)入integration_tests/delivery目錄，并從docker compose文件啟動(dòng)演示集群。

cd risingwave/integration_tests/delivery
docker compose up -d

確保所有容器都已啟動(dòng)并運(yùn)行。

步驟2：安裝Dash和Psycopg2庫(kù)

要安裝Dash，也可以參考網(wǎng)站上的Dash安裝指南?；旧?，需要通過(guò)運(yùn)行以下pip install命令來(lái)安裝兩個(gè)庫(kù)(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來(lái)與RisingWave流數(shù)據(jù)庫(kù)交互：

pip install psycopg2-binary

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

要使用RisingWave獲取實(shí)時(shí)數(shù)據(jù)，首先需要設(shè)置一個(gè)數(shù)據(jù)源。在演示項(xiàng)目中，Kafka應(yīng)該被定義為數(shù)據(jù)源。將創(chuàng)建一個(gè)名為create-a-source.py的新文件，與連接到RisingWave的Python腳本的integration_tests/delivery目錄相同，并創(chuàng)建一個(gè)表來(lái)消費(fèi)和持久化delivery_orders Kafka主題?？梢院?jiǎn)單地將以下代碼復(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)建文件之后，運(yùn)行python create-a-source.py，它將在RisingWave中創(chuàng)建源表。

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

接下來(lái)，創(chuàng)建一個(gè)新的物化視圖，類似于創(chuàng)建表的方式。在此創(chuàng)建一個(gè)名為create-a-materialized-view.py的新文件，并使用Psycopg2庫(kù)運(yùn)行SQL查詢。也可以將上面的最后兩個(gè)步驟合并到一個(gè)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查詢實(shí)時(shí)計(jì)算特定餐廳在過(guò)去15分鐘內(nèi)創(chuàng)建的訂單總數(shù)，并將結(jié)果緩存到物化視圖中。如果發(fā)生任何數(shù)據(jù)更改或新的Kafka主題，RisingWave會(huì)自動(dòng)增加和更新物化視圖的結(jié)果。一旦設(shè)置了數(shù)據(jù)源，物化視圖，就可以開始攝取數(shù)據(jù)，并使用Dash將這些數(shù)據(jù)可視化。

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

現(xiàn)在構(gòu)建Dash應(yīng)用程序來(lái)查詢和可視化RisingWave中的物化視圖內(nèi)容。可以在20分鐘內(nèi)跟隨教程Dash來(lái)了解Dash的基本構(gòu)建塊。這一示例的應(yīng)用程序代碼以表格和圖形兩種格式顯示餐廳訂單數(shù)據(jù)。請(qǐng)參閱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'）相對(duì)應(yīng)，行與物化中的數(shù)據(jù)相對(duì)應(yīng)。

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

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

總結(jié)

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

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