파이썬, 대시 및 RisingWave를 사용하여 실시간 데이터 시각화

튜토리얼
Kafka Pandas SQL

실시간 데이터는 기업이 신속한 의사결정을 할 수 있게 해주는데 중요합니다. 이 데이터를 시각적으로 보는 것은 의사결정을 더 빠르게 할 수 있습니다. 다양한 데이터 앱이나 대시보드를 사용하여 데이터의 시각적 표현을 만들 수 있습니다. Dash는 대화형 차트, 그래프, 테이블 및 기타 UI 요소를 만들기 위한 다양한 기본 제공 구성 요소를 제공하는 오픈 소스 Python 라이브러리입니다. RisingWave는 실시간 데이터 처리를 위한 SQL 기반 스트리밍 데이터베이스입니다. 이 기사에서는 Python, Dash, RisingWave를 사용하여 실시간 데이터의 시각화를 만드는 방법을 설명할 것입니다.

실시간 데이터 시각화 방법

실시간 데이터는 수집되는 즉시 생성되고 처리되는 데이터라는 것을 알고 있습니다. 소스는 Postgres나 MySQL과 같은 전형적인 데이터베이스나 Kafka와 같은 메시지 브로커일 수 있습니다. 실시간 데이터 시각화에는 몇 가지 단계가 있습니다: 먼저 데이터 수집, 그다음 처리, 마지막으로 이 데이터를 대시보드에 표시합니다.

주문 배달 데이터의 경우, 이 데이터를 실시간으로 시각화하면 식당이나 배달 서비스의 성과에 대한 귀중한 통찰력을 제공할 수 있습니다. 예를 들어, 실시간 데이터를 사용하여 주문이 배달되는 데 걸리는 시간을 모니터링하고, 배달 과정의 병목 현상을 식별하고, 시간이 지남에 따라 주문량의 변화를 추적할 수 있습니다. 끊임없이 변화하는 데이터를 다룰 때 모든 일이 일어나고 있는지 추적하고 패턴이나 추세를 식별하기가 어려울 수 있습니다. Dash 및 RisingWave과 같은 무료 도구를 사용하면 이 끊임없이 변화하는 데이터를 탐색하고 분석할 수 있는 대화형 시각화를 만들 수 있습니다.

데이터 작업에 대해 생각할 때 첫 번째 프로그래밍 언어는 Python일 수 있습니다. 라이브러리가 다양하기 때문입니다. Dash는 그 중 하나로, 단순한 Python 코드만으로 풍부하고 사용자 정의 가능한 사용자 인터페이스를 가진 데이터 앱을 만들 수 있습니다. Dash는 인기 있는 웹 개발 도구인 Flask, Plotly.js 및 React.js 위에 구축되어 있으므로 HTML, CSS 또는 기타 JavaScript 프레임워크를 알 필요가 없습니다.

RisingWave을 사용하면 다양한 소스에서 데이터 스트림을 소비하고 복잡한 쿼리를 위해 최적화된 물리적 뷰를 생성하고 SQL을 사용하여 실시간 데이터 쿼리할 수 있습니다. RisingWave이 PostgreSQL과 호환되므로 psycopg2(Python의 PostgreSQL 클라이언트 라이브러리) 드라이버를 사용하여 RisingWave에 연결하고 쿼리 작업을 수행할 수 있습니다. 다음 섹션에서 확인해보세요.

Visualize Order Delivery Data Demo

이 데모 튜토리얼에서는 Docker Compose를 사용하여 필요한 모든 설정이 완료되었다고 가정하고, RisingWave 데모를 포함한 GitHub 리포지토리를 활용합니다. 다른 방법으로 RisingWave 실행을 원하신다면 공식 웹사이트를 확인하세요. 우리는 음식 배달 웹사이트에서 발생한 모든 주문에 대한 이벤트를 포함하는 delivery_orders라는 Kafka 토픽이 있습니다. 각 이벤트에는 주문 ID, 음식점 ID, 배달 상태와 같은 주문에 대한 정보가 포함되어 있습니다. 부하 생성기(Datagen이라는 Python 스크립트)는 랜덤 모의 데이터를 지속적으로 생성하고 이를 Kafka 토픽에 스트리밍합니다. 실제로는 이 모의 데이터를 웹 앱이나 백엔드 서비스에서 오는 데이터로 대체할 수 있습니다.

시작하기 전에

이 튜토리얼을 완료하려면 다음이 필요합니다:

  • Docker와 Docker Compose가 환경에 설치되어 있는지 확인하세요.
  • PostgreSQL 대화형 터미널인 psql이 환경에 설치되어 있는지 확인하세요. 자세한 지침은 PostgreSQL 다운로드를 참조하세요.
  • 운영 체제에 맞는 Python 3를 다운로드하고 설치하세요. pip 명령어가 자동으로 설치됩니다.

저는 Windows OS, 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

또한 RisingWave 스트리밍 데이터베이스와 상호작용하기 위해 psycopg2도 설치해야 합니다:

pip install psycopg2-binary

3단계: 데이터 소스 생성

RisingWave로 실시간 데이터를 수집하려면 먼저 데이터 소스를 설정해야 합니다. 데모 프로젝트에서 Kafka는 데이터 소스로 정의되어야 합니다. 새 파일인 create-a-source.py를 만들어서 integration_tests/delivery 디렉토리에 있는 Python 스크립트와 동일한 위치에 RisingWave에 연결하고, delivery_orders Kafka 토픽을 소비하고 지속시킬 테이블을 생성할 것입니다. 아래의 코드를 새 파일에 간단히 복사 붙여넣기 하시면 됩니다.

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 앱 구축

이제 RisingWave에 있는 물리적 뷰 콘텐츠를 쿼리하고 시각화하기 위해 Dash 앱을 구축합니다. Dash의 기본 빌딩 블록을 이해하기 위해 20분 내 대시 튜토리얼을 따를 수 있습니다. 예제 애플리케이션 코드는 테이블 및 그래프 형식으로 레스토랑 주문 데이터를 표시합니다. 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는 복잡한 UI와 데이터 시각화 기능이 필요한 데이터 분석 뷰를 만들기 위한 강력한 도구로, Python 프로그래밍 언어의 단순성과 우아함을 사용합니다. RisingWave 스트리밍 데이터베이스와 함께 사용할 때, 실시간 데이터에 대한 통찰력을 얻을 수 있어 보다 정보에 기반한 의사 결정과 성과 최적화를 위한 조치를 취할 수 있습니다.

관련 자료

Source:
https://dzone.com/articles/visualize-real-time-data-with-python-dash-and-risi