تصور البيانات في الوقت الفعلي باستخدام بايثون، داش، وريزينغ ويف

البيانات في الوقت الفعلي مهمة للشركات لاتخاذ قرارات سريعة. رؤية هذه البيانات بصرياً يمكن أن تساعد في اتخاذ القرارات بشكل أسرع. يمكننا إنشاء تمثيلات بصرية للبيانات باستخدام تطبيقات بيانات مختلفة أو لوحات تحكم. تمثيل هي مكتبة Python مفتوحة المصدر توفر مجموعة واسعة من المكونات المدمجة لإنشاء رسوم بيانية تفاعلية، جداول، وعناصر UI الأخرى. تضليع الموجة قاعدة بيانات تدفق يعتمد على SQL لمعالجة البيانات في الوقت الفعلي. يوضح هذا المقال كيفية استخدام بيثون، تمثيل، وتضليع الموجة لإنشاء تصورات للبيانات في الوقت الفعلي.

كيفية تصور البيانات في الوقت الفعلي

نعلم أن البيانات في الوقت الفعلي هي بيانات تُولد وتُعالج على الفور، حيث تُجمع من مصادر بيانات مختلفة. يمكن للمصادر أن تكون قواعد بيانات نموذجية مثل بوستجريس أو ماي إس أي، ووكالات نقل الرسائل مثل كافكا. تتألف تصور البيانات في الوقت الفعلي من عدة خطوات: أولاً، نقوم باستيعابها، ثم معالجتها، وأخيراً عرض هذه البيانات في لوحة تحكم.

في حالة ملفات تعريف التسليم، يمكن أن يوفر تصور هذه البيانات في الوقت الفعلي رؤى قيمة في أداء مطعم أو خدمة تسليم. على سبيل المثال، يمكننا استخدام البيانات الفورية لمراقبة مدة الوقت التي يستغرقها تسليم الطلبات، تحديد العقبات في عملية التسليم، وتتبع تغيرات حجم الطلبات مع مرور الوقت. عند التعامل مع البيانات التي تتغير باستمرار، يمكن أن يكون من الصعب تتبع كل شيء يحدث وتحديد الأنماط أو الاتجاهات. باستخدام أدوات مجانية مثل Dash و RisingWave، يمكننا إنشاء تصورات تفاعلية تسمح لنا باستكشاف وتحليل هذه البيانات المتغيرة باستمرار.

عندما يتعلق الأمر بالعمل مع البيانات، قد يكون أول لغة برمجة تفكر فيها هي Python، لأن لديها مجموعة من المكتبات. Dash هي واحدة منها تسمح لنا بإنشاء تطبيق بيانات بواجهة مستخدم غنية وقابلة للتخصيص باستخدام لغة برمجة Python فقط. تم إنشاء Dash على رأس Flask، Plotly.js، و React.js، وهي أدوات تطوير الويب الشهيرة لذا لا تحتاج إلى معرفة HTML، CSS، أو إطارات JavaScript أخرى.

باستخدام RisingWave، يمكننا استهلاك تيارات البيانات من مصادر مختلفة، إنشاء عروض مادية محسنة للاستعلامات المعقدة، واستعلام عن البيانات الفورية باستخدام SQL. حيث أن RisingWave متوافق عبر الخط مع PostgreSQL، يمكننا استخدام برنامج psycopg2 (مكتبة عميل PostgreSQL في Python) للاتصال بـ RisingWave، وإجراء عمليات الاستعلام. انظر في القسم التالي.

تجسيد توصيل بيانات الطلب للعرض التوضيحي

في البرنامج التعليمي للعرض التوضيحي، سنستفيد من مستودع GitHub مع العروض التوضيحية لـ RisingWave حيث نفترض أن كل الأشياء اللازمة معلية استخدام Docker Compose. يمكنك التحقق من طرق أخرى لـتشغيل RisingWave على الموقع الرسمي. لدينا موضوع Kafka باسم delivery_orders الذي يحتوي على أحداث لكل طلب تم إجراؤه على موقع توصيل الطعام. يشمل كل حدث معلومات حول الطلب، مثل معرف الطلب، معرف المطعم، و حالة التوصيل. محول العبء (سكربت Python يسمى Datagen) يحاكي توليد بيانات شبه عشوائية باستمرار ويسيرها إلى مواضع Kafka. في الواقع، يمكن استبدال هذه البيانات المصغرة ببيانات تأتي من تطبيقك الإلكتروني أو خدمة الخلفية.

قبل البدء

لإكمال هذا البرنامج التعليمي، تحتاج إلى ما يلي:

• تأكد من تثبيت Docker و Docker Compose في بيئتك.
• تأكد من تثبيت المحطة التفاعلية PostgreSQL psql في بيئتك. للحصول على تعليمات مفصلة، انظر تنزيل PostgreSQL.
• قم بتنزيل وتثبيت Python 3 لنظام التشغيل الخاص بك. سيتم تثبيت pip الأمر تلقائيًا.

لقد اختبرت العرض التوضيحي على نظام التشغيل 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 على الموقع. بشكل أساسي، نحتاج لتثبيت مكتبتين (Dash نفسها وPandas) عن طريق تشغيل الأمر pip install التالي:

# 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، نفس مجلد integration_tests/delivery الذي يحتوي على سيرفت بايثون حيث نتصل بـ RisingWave، وننشئ جدولًا لاستهلاك والاحتفاظ بمواضع Kafka delivery_orders. يمكنك ببساطة نسخ ولصق الكود التالي في الملف الجديد.

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 ونقوم بتشغيل عملية استعلام SQL باستخدام مكتبة psycopg2. من الممكن أيضًا دمج الخطوتين الأخيرتين في ملف 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. يمكنك اتباع البرنامج التعليمي Dash في 20 دقيقة لفهم اللبنات الأساسية لـ Dash. رمز تطبيقنا العيني يعرض بيانات طلبات المطاعم بصيغة جدول ورسم بياني. انظر الكود البرمجي الآتي في 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)

هذه المقتطف البرمجي يسترد البيانات من واجهة المعاينة restaurant_orders_view باستخدام بانداس ويعرضها في جدول باستخدام dash_table.DataTable ومخطط شريطي باستخدام dcc.Graph. يحتوي الجدول والمخطط الشريطي على أعمدة تتوافق مع الأعمدة في واجهة المعاينة (‘window_start’, ‘total_order’, و ‘restaurant_id’) وصفوف تتوافق مع البيانات في واجهة المعاينة.

الخطوة 6: عرض النتائج

يمكنك تشغيل التطبيق عن طريق تشغيل السكريبت dash-example.py أعلاه والتنقل إلى http://localhost:8050/ في متصفح الويب الخاص بك (سوف تتلقى رسالة في المحطة النصية تخبرك بالذهاب إلى هذا الرابط).

ملخص

المجمل، Dash هو أداة قوية لإنشاء عرض البيانات التحليلية التي تتطلب واجهات معقدة وقدرات تصور البيانات، وكلها ببساطة وأناقة لغة البرمجة Python. عندما نستخدمه مع قاعدة بيانات البث المتسلسل RisingWave، نكتسب رؤى في البيانات في الوقت الحقيقي التي يمكن أن تساعدنا في اتخاذ قرارات أكثر استنارة واتخاذ إجراءات لتحسين الأداء.

الموارد ذات الصلة

• تحليل البيانات في الوقت الحقيقي باستخدام Apache Superset، Redpanda، وRisingWave
• كيفية مراقبة مقاييس البث المباشر