Phân tích dữ liệu thời gian thực với Kafka và Python
2024-03-25 — QuantTrade
Giới thiệu
Trong thời đại bùng nổ dữ liệu, khả năng thu thập, xử lý và phân tích dữ liệu theo thời gian thực đã trở thành một yêu cầu quan trọng đối với nhiều tổ chức. Từ việc giám sát hệ thống CNTT, phát hiện gian lận, đến việc đưa ra quyết định kinh doanh nhanh chóng, phân tích dữ liệu thời gian thực mang lại giá trị to lớn.
Apache Kafka kết hợp với Python tạo nên một bộ công cụ mạnh mẽ để xây dựng các hệ thống phân tích dữ liệu thời gian thực. Kafka cung cấp nền tảng truyền tải dữ liệu phân tán, khả năng mở rộng cao, trong khi Python mang đến hệ sinh thái phong phú các thư viện phân tích dữ liệu và học máy.
Bài viết này sẽ hướng dẫn chi tiết cách kết hợp Kafka và Python để xây dựng một hệ thống phân tích dữ liệu thời gian thực.
1. Tổng quan về Apache Kafka
1.1 Kafka là gì?
Apache Kafka là một nền tảng xử lý luồng phân tán (distributed streaming platform) được phát triển ban đầu bởi LinkedIn và hiện đang là một dự án mã nguồn mở của Apache Software Foundation. Kafka được thiết kế để xử lý dữ liệu thời gian thực với lưu lượng cao, độ trễ thấp.
1.2 Kiến trúc cơ bản của Kafka
Kafka bao gồm các thành phần chính sau:
- Broker: Các máy chủ Kafka, lưu trữ và phục vụ dữ liệu
- Topic: Kênh dữ liệu nơi các thông điệp được đăng tải
- Partition: Các phần của topic, giúp phân tán và song song hóa
- Producer: Ứng dụng gửi dữ liệu đến Kafka
- Consumer: Ứng dụng đọc dữ liệu từ Kafka
- Consumer Group: Nhóm consumer phối hợp để xử lý dữ liệu
- ZooKeeper: Quản lý cấu hình và điều phối (trong các phiên bản mới có thể không cần)
1.3 Ưu điểm của Kafka trong phân tích thời gian thực
- Khả năng mở rộng: Có thể xử lý hàng triệu thông điệp mỗi giây
- Độ tin cậy cao: Dữ liệu được sao chép để đảm bảo không mất mát
- Độ trễ thấp: Đáp ứng thời gian thực với độ trễ chỉ vài mili giây
- Lưu trữ dữ liệu: Lưu trữ dữ liệu trong thời gian có thể cấu hình
- Khả năng phục hồi: Thiết kế phân tán giúp chịu lỗi tốt
2. Python trong phân tích dữ liệu thời gian thực
2.1 Lợi thế của Python
Python đã trở thành ngôn ngữ hàng đầu cho phân tích dữ liệu và học máy với những lợi thế:
- Dễ học và sử dụng: Cú pháp rõ ràng, đơn giản
- Hệ sinh thái phong phú: Hàng nghìn thư viện chuyên dụng
- Cộng đồng lớn: Hỗ trợ và tài liệu phong phú
- Tích hợp tốt: Dễ dàng kết hợp với các công nghệ khác
- Năng suất cao: Phát triển nhanh, vòng đời ngắn
2.2 Các thư viện Python cho phân tích dữ liệu thời gian thực
- kafka-python, confluent-kafka: Thư viện kết nối Kafka
- pandas, NumPy: Xử lý và phân tích dữ liệu
- scikit-learn, PyTorch, TensorFlow: Học máy và deep learning
- Spark (PySpark): Xử lý dữ liệu phân tán
- Faust: Framework xử lý luồng dữ liệu
- Streamlit, Dash, Bokeh: Trực quan hóa dữ liệu thời gian thực
- Flask, FastAPI: Xây dựng API và dịch vụ web
3. Thiết lập môi trường Kafka và Python
3.1 Cài đặt Kafka
Có nhiều cách để thiết lập Kafka:
Sử dụng Docker
# Tạo file docker-compose.yml
cat > docker-compose.yml << EOF
version: '3'
services:
zookeeper:
image: confluentinc/cp-zookeeper:latest
environment:
ZOOKEEPER_CLIENT_PORT: 2181
kafka:
image: confluentinc/cp-kafka:latest
depends_on:
- zookeeper
ports:
- "9092:9092"
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
EOF
# Khởi động Kafka và ZooKeeper
docker-compose up -d
Cài đặt trực tiếp
# Tải và giải nén Kafka
wget https://downloads.apache.org/kafka/3.3.2/kafka_2.13-3.3.2.tgz
tar -xzf kafka_2.13-3.3.2.tgz
cd kafka_2.13-3.3.2
# Khởi động ZooKeeper
bin/zookeeper-server-start.sh config/zookeeper.properties &
# Khởi động Kafka
bin/kafka-server-start.sh config/server.properties &
3.2 Cài đặt thư viện Python
# Cài đặt thư viện cần thiết
pip install confluent-kafka pandas numpy matplotlib streamlit scikit-learn
3.3 Tạo môi trường ảo Python
# Tạo môi trường ảo
python -m venv kafka-analytics
source kafka-analytics/bin/activate # Linux/Mac
# hoặc
kafka-analytics\Scripts\activate # Windows
4. Tạo Producer - Thu thập dữ liệu thời gian thực
4.1 Producer đơn giản với confluent-kafka
from confluent_kafka import Producer
import json
import random
import time
from datetime import datetime
# Cấu hình Producer
conf = {'bootstrap.servers': 'localhost:9092'}
producer = Producer(conf)
# Hàm callback khi gửi thông điệp
def delivery_report(err, msg):
if err is not None:
print(f'Lỗi gửi thông điệp: {err}')
else:
print(f'Thông điệp đã gửi đến {msg.topic()} [{msg.partition()}] tại offset {msg.offset()}')
# Tạo dữ liệu giả lập cảm biến IoT
def generate_sensor_data():
return {
'sensor_id': random.randint(1, 10),
'temperature': round(random.uniform(20, 35), 2),
'humidity': round(random.uniform(30, 90), 2),
'pressure': round(random.uniform(980, 1050), 2),
'timestamp': datetime.now().isoformat()
}
# Topic để gửi dữ liệu
topic = 'sensor-data'
# Gửi dữ liệu liên tục
try:
while True:
# Tạo dữ liệu cảm biến
sensor_data = generate_sensor_data()
# Chuyển đổi dữ liệu sang JSON
message = json.dumps(sensor_data)
# Gửi thông điệp đến Kafka
producer.produce(topic, key=str(sensor_data['sensor_id']), value=message, callback=delivery_report)
# Đảm bảo tất cả thông điệp được gửi
producer.poll(0)
print(f"Đã gửi: {message}")
time.sleep(1) # Gửi dữ liệu mỗi giây
except KeyboardInterrupt:
print("Đã dừng producer")
finally:
# Đảm bảo tất cả thông điệp còn lại được gửi
producer.flush()
4.2 Tích hợp với các nguồn dữ liệu thực
Thu thập dữ liệu từ API
import requests
from confluent_kafka import Producer
import json
import time
# Cấu hình Producer
conf = {'bootstrap.servers': 'localhost:9092'}
producer = Producer(conf)
# Topic để gửi dữ liệu
topic = 'market-data'
# Hàm lấy dữ liệu từ API
def get_market_data():
# Ví dụ với API dữ liệu thị trường
response = requests.get('https://api.example.com/market-data')
if response.status_code == 200:
return response.json()
else:
print(f"Lỗi khi lấy dữ liệu: {response.status_code}")
return None
# Gửi dữ liệu liên tục
try:
while True:
# Lấy dữ liệu
data = get_market_data()
if data:
# Chuyển đổi dữ liệu sang JSON
message = json.dumps(data)
# Gửi thông điệp đến Kafka
producer.produce(topic, value=message)
# Đảm bảo thông điệp được gửi
producer.poll(0)
print(f"Đã gửi dữ liệu thị trường")
time.sleep(60) # Cập nhật mỗi phút
except KeyboardInterrupt:
print("Đã dừng producer")
finally:
producer.flush()
Thu thập dữ liệu từ file logs
import os
import time
from confluent_kafka import Producer
import json
# Cấu hình Producer
conf = {'bootstrap.servers': 'localhost:9092'}
producer = Producer(conf)
# Topic để gửi dữ liệu
topic = 'app-logs'
# Hàm theo dõi file log
def tail_log_file(file_path):
with open(file_path, 'r') as file:
# Di chuyển đến cuối file
file.seek(0, os.SEEK_END)
while True:
# Đọc dòng mới
line = file.readline()
# Nếu không có dòng mới, đợi và thử lại
if not line:
time.sleep(0.1)
continue
# Xử lý dòng log
try:
# Giả sử log ở định dạng JSON
log_data = json.loads(line)
# Gửi dữ liệu đến Kafka
producer.produce(topic, value=line.strip())
producer.poll(0)
print(f"Đã gửi log: {line.strip()}")
except json.JSONDecodeError:
# Nếu không phải JSON, gửi nguyên dòng text
producer.produce(topic, value=line.strip())
producer.poll(0)
print(f"Đã gửi log dạng text: {line.strip()}")
# Bắt đầu theo dõi file log
try:
log_file = '/var/log/application.log' # Thay đổi đường dẫn phù hợp
tail_log_file(log_file)
except KeyboardInterrupt:
print("Đã dừng producer")
finally:
producer.flush()
5. Tạo Consumer - Xử lý dữ liệu thời gian thực
5.1 Consumer đơn giản với confluent-kafka
from confluent_kafka import Consumer
import json
import matplotlib.pyplot as plt
import pandas as pd
from collections import deque
import threading
# Cấu hình Consumer
conf = {
'bootstrap.servers': 'localhost:9092',
'group.id': 'sensor-analytics',
'auto.offset.reset': 'earliest'
}
consumer = Consumer(conf)
# Đăng ký topic
topic = 'sensor-data'
consumer.subscribe([topic])
# Tạo cấu trúc dữ liệu để lưu trữ
temperature_data = deque(maxlen=100) # Giữ 100 điểm dữ liệu gần nhất
humidity_data = deque(maxlen=100)
timestamps = deque(maxlen=100)
# Hàm cập nhật biểu đồ thời gian thực
def update_plot():
plt.ion() # Chế độ tương tác
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 8))
temp_line, = ax1.plot([], [], 'r-')
hum_line, = ax2.plot([], [], 'b-')
ax1.set_title('Nhiệt độ theo thời gian thực')
ax1.set_ylabel('Nhiệt độ (°C)')
ax1.set_ylim(15, 40)
ax2.set_title('Độ ẩm theo thời gian thực')
ax2.set_ylabel('Độ ẩm (%)')
ax2.set_xlabel('Thời gian')
ax2.set_ylim(20, 100)
plt.tight_layout()
while True:
if len(timestamps) > 0:
temp_line.set_data(range(len(temperature_data)), list(temperature_data))
hum_line.set_data(range(len(humidity_data)), list(humidity_data))
ax1.set_xlim(0, len(temperature_data))
ax2.set_xlim(0, len(humidity_data))
# Cập nhật nhãn thời gian
if len(timestamps) >= 5:
positions = [0, len(timestamps)//4, len(timestamps)//2, 3*len(timestamps)//4, len(timestamps)-1]
labels = [list(timestamps)[i].split('T')[1].split('.')[0] for i in positions]
ax2.set_xticks(positions)
ax2.set_xticklabels(labels, rotation=45)
fig.canvas.draw()
fig.canvas.flush_events()
plt.pause(0.1)
# Khởi động luồng cập nhật biểu đồ
plot_thread = threading.Thread(target=update_plot)
plot_thread.daemon = True
plot_thread.start()
# Xử lý thông điệp
print(f"Bắt đầu lắng nghe topic {topic}...")
try:
while True:
# Poll để nhận thông điệp
msg = consumer.poll(1.0)
if msg is None:
continue
if msg.error():
print(f"Lỗi Consumer: {msg.error()}")
continue
# Xử lý thông điệp
try:
value = msg.value().decode('utf-8')
data = json.loads(value)
# Lưu dữ liệu
temperature_data.append(data['temperature'])
humidity_data.append(data['humidity'])
timestamps.append(data['timestamp'])
print(f"Đã nhận: Cảm biến {data['sensor_id']}, Nhiệt độ: {data['temperature']}°C, Độ ẩm: {data['humidity']}%")
except Exception as e:
print(f"Lỗi khi xử lý thông điệp: {e}")
except KeyboardInterrupt:
print("Đã dừng consumer")
finally:
consumer.close()
5.2 Phân tích dữ liệu với pandas và NumPy
from confluent_kafka import Consumer
import json
import pandas as pd
import numpy as np
from datetime import datetime
import time
# Cấu hình Consumer
conf = {
'bootstrap.servers': 'localhost:9092',
'group.id': 'sensor-analytics-batch',
'auto.offset.reset': 'earliest'
}
consumer = Consumer(conf)
# Đăng ký topic
topic = 'sensor-data'
consumer.subscribe([topic])
# Kích thước batch và window phân tích
BATCH_SIZE = 100
WINDOW_SIZE = 10 # Phân tích cửa sổ trượt mỗi 10 phút
# Lưu trữ dữ liệu
data_points = []
# Hàm phân tích batch dữ liệu
def analyze_batch(batch_df):
# Thống kê tổng quan
print(f"\n=== Phân tích batch có {len(batch_df)} điểm dữ liệu ===")
# Thống kê theo cảm biến
sensor_stats = batch_df.groupby('sensor_id').agg({
'temperature': ['mean', 'min', 'max', 'std'],
'humidity': ['mean', 'min', 'max', 'std']
})
print("\nThống kê theo cảm biến:")
print(sensor_stats)
# Phát hiện bất thường
# Ví dụ: Nhiệt độ vượt ngưỡng
temp_anomalies = batch_df[batch_df['temperature'] > 30]
if not temp_anomalies.empty:
print("\nPhát hiện bất thường nhiệt độ (>30°C):")
print(temp_anomalies[['sensor_id', 'temperature', 'timestamp']])
# Phân tích tương quan
correlation = batch_df[['temperature', 'humidity', 'pressure']].corr()
print("\nMa trận tương quan:")
print(correlation)
# Phân tích xu hướng
batch_df['timestamp'] = pd.to_datetime(batch_df['timestamp'])
batch_df.set_index('timestamp', inplace=True)
# Lấy phân tích theo thời gian
if len(batch_df) >= WINDOW_SIZE:
# Phân tích cửa sổ trượt
rolling_temp = batch_df['temperature'].rolling(window=WINDOW_SIZE).mean()
print("\nNhiệt độ trung bình cửa sổ trượt:")
print(rolling_temp.tail())
# Xử lý thông điệp
print(f"Bắt đầu thu thập dữ liệu từ topic {topic}...")
try:
while True:
# Poll để nhận thông điệp
msg = consumer.poll(1.0)
if msg is None:
continue
if msg.error():
print(f"Lỗi Consumer: {msg.error()}")
continue
# Xử lý thông điệp
try:
value = msg.value().decode('utf-8')
data = json.loads(value)
# Thêm vào danh sách dữ liệu
data_points.append(data)
# Khi đủ kích thước batch, thực hiện phân tích
if len(data_points) >= BATCH_SIZE:
# Chuyển danh sách dữ liệu thành DataFrame
batch_df = pd.DataFrame(data_points)
# Phân tích batch
analyze_batch(batch_df)
# Giữ lại 20% dữ liệu gần nhất để phân tích cửa sổ trượt
keep_size = int(BATCH_SIZE * 0.2)
data_points = data_points[-keep_size:]
print(f"\nĐã xử lý batch. Giữ lại {keep_size} điểm dữ liệu cho cửa sổ trượt.")
except Exception as e:
print(f"Lỗi khi xử lý thông điệp: {e}")
except KeyboardInterrupt:
print("Đã dừng consumer")
finally:
consumer.close()
5.3 Áp dụng học máy cho dữ liệu thời gian thực
from confluent_kafka import Consumer
import json
import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import joblib
import os
from datetime import datetime
# Cấu hình Consumer
conf = {
'bootstrap.servers': 'localhost:9092',
'group.id': 'sensor-anomaly-detection',
'auto.offset.reset': 'earliest'
}
consumer = Consumer(conf)
# Đăng ký topic
topic = 'sensor-data'
consumer.subscribe([topic])
# Tham số
MODEL_PATH = 'models/isolation_forest.joblib'
BATCH_SIZE = 200 # Kích thước lô để huấn luyện/cập nhật mô hình
FEATURES = ['temperature', 'humidity', 'pressure'] # Đặc trưng cho phát hiện bất thường
# Biến để lưu trữ dữ liệu và mô hình
data_points = []
scaler = StandardScaler()
model = None
# Tạo thư mục lưu mô hình nếu chưa tồn tại
os.makedirs(os.path.dirname(MODEL_PATH), exist_ok=True)
# Kiểm tra và tải mô hình đã huấn luyện (nếu có)
if os.path.exists(MODEL_PATH):
print(f"Tải mô hình từ {MODEL_PATH}")
model = joblib.load(MODEL_PATH)
print("Đã tải mô hình thành công!")
# Hàm huấn luyện/cập nhật mô hình
def train_model(data):
global model, scaler
print("Huấn luyện mô hình phát hiện bất thường...")
# Chuyển đổi dữ liệu
X = data[FEATURES].copy()
# Chuẩn hóa dữ liệu
scaler.fit(X)
X_scaled = scaler.transform(X)
# Huấn luyện mô hình Isolation Forest
if model is None:
model = IsolationForest(n_estimators=100, contamination=0.05, random_state=42)
# Huấn luyện mô hình
model.fit(X_scaled)
# Lưu mô hình
joblib.dump(model, MODEL_PATH)
print("Đã huấn luyện và lưu mô hình thành công!")
# Hàm phát hiện bất thường
def detect_anomalies(data):
if model is None:
print("Chưa có mô hình. Cần thu thập đủ dữ liệu để huấn luyện.")
return None
# Chuẩn bị dữ liệu
X = data[FEATURES].copy()
# Chuẩn hóa dữ liệu
X_scaled = scaler.transform(X)
# Dự đoán bất thường (-1 là bất thường, 1 là bình thường)
predictions = model.predict(X_scaled)
anomaly_scores = model.decision_function(X_scaled)
# Thêm kết quả vào dữ liệu
data['anomaly'] = predictions
data['anomaly_score'] = anomaly_scores
# Lấy các bất thường
anomalies = data[data['anomaly'] == -1]
return anomalies
# Xử lý thông điệp
print(f"Bắt đầu phát hiện bất thường từ topic {topic}...")
try:
while True:
# Poll để nhận thông điệp
msg = consumer.poll(1.0)
if msg is None:
continue
if msg.error():
print(f"Lỗi Consumer: {msg.error()}")
continue
# Xử lý thông điệp
try:
value = msg.value().decode('utf-8')
data = json.loads(value)
# Thêm vào danh sách dữ liệu
data_points.append(data)
# Khi có đủ dữ liệu để huấn luyện
if len(data_points) >= BATCH_SIZE and model is None:
batch_df = pd.DataFrame(data_points)
train_model(batch_df)
# Khi đã có mô hình và có dữ liệu mới
elif model is not None and len(data_points) % 10 == 0: # Kiểm tra mỗi 10 điểm dữ liệu
# Phát hiện bất thường cho dữ liệu mới nhất
recent_data = pd.DataFrame(data_points[-10:])
anomalies = detect_anomalies(recent_data)
if anomalies is not None and not anomalies.empty:
print(f"\n=== Phát hiện {len(anomalies)} bất thường ===")
for idx, row in anomalies.iterrows():
print(f"Cảm biến {row['sensor_id']} lúc {row['timestamp']}")
print(f" Nhiệt độ: {row['temperature']}°C, Độ ẩm: {row['humidity']}%, Áp suất: {row['pressure']} hPa")
print(f" Điểm bất thường: {row['anomaly_score']:.4f}")
# Cập nhật mô hình định kỳ
if len(data_points) >= BATCH_SIZE * 5: # Cập nhật sau mỗi 5 batch
batch_df = pd.DataFrame(data_points)
train_model(batch_df)
# Giữ lại 20% dữ liệu gần nhất
keep_size = int(BATCH_SIZE)
data_points = data_points[-keep_size:]
except Exception as e:
print(f"Lỗi khi xử lý thông điệp: {e}")
except KeyboardInterrupt:
print("Đã dừng consumer")
finally:
consumer.close()
6. Xây dựng ứng dụng Streamlit để trực quan hóa dữ liệu thời gian thực
6.1 Ứng dụng Streamlit cơ bản
# Lưu file này là app.py
import streamlit as st
import pandas as pd
import numpy as np
import altair as alt
import json
from confluent_kafka import Consumer
from datetime import datetime
import time
import threading
import queue
# Khởi tạo giao diện Streamlit
st.title('Phân tích dữ liệu cảm biến thời gian thực')
st.write('Dashboard hiển thị dữ liệu cảm biến từ Kafka theo thời gian thực')
# Tạo placeholder cho biểu đồ
chart_placeholder = st.empty()
metrics_placeholder = st.empty()
table_placeholder = st.empty()
# Tạo hàng đợi để truyền dữ liệu từ Consumer thread
data_queue = queue.Queue()
# Dữ liệu cảm biến
sensor_data = []
max_data_points = 100 # Số điểm dữ liệu tối đa hiển thị
# Hàm Consumer Kafka chạy trong thread riêng
def kafka_consumer():
# Cấu hình Consumer
conf = {
'bootstrap.servers': 'localhost:9092',
'group.id': 'streamlit-dashboard',
'auto.offset.reset': 'latest'
}
consumer = Consumer(conf)
consumer.subscribe(['sensor-data'])
try:
while True:
# Poll để nhận thông điệp
msg = consumer.poll(1.0)
if msg is None:
continue
if msg.error():
print(f"Lỗi Consumer: {msg.error()}")
continue
# Xử lý thông điệp
try:
value = msg.value().decode('utf-8')
data = json.loads(value)
# Thêm timestamp dễ đọc
data['readable_time'] = datetime.fromisoformat(data['timestamp']).strftime('%H:%M:%S')
# Đưa dữ liệu vào hàng đợi
data_queue.put(data)
except Exception as e:
print(f"Lỗi khi xử lý thông điệp: {e}")
except Exception as e:
print(f"Lỗi trong consumer thread: {e}")
finally:
consumer.close()
# Khởi động Consumer thread
consumer_thread = threading.Thread(target=kafka_consumer)
consumer_thread.daemon = True
consumer_thread.start()
# Hiển thị dữ liệu
try:
while True:
# Nhận dữ liệu từ hàng đợi
try:
while not data_queue.empty():
data = data_queue.get(block=False)
sensor_data.append(data)
# Giới hạn số lượng điểm dữ liệu
if len(sensor_data) > max_data_points:
sensor_data.pop(0)
except queue.Empty:
pass
if sensor_data:
# Tạo DataFrame
df = pd.DataFrame(sensor_data)
# Hiển thị metrics
with metrics_placeholder.container():
cols = st.columns(4)
cols[0].metric("Nhiệt độ trung bình", f"{df['temperature'].mean():.2f}°C",
f"{df['temperature'].iloc[-1] - df['temperature'].iloc[0]:.2f}°C")
cols[1].metric("Độ ẩm trung bình", f"{df['humidity'].mean():.2f}%",
f"{df['humidity'].iloc[-1] - df['humidity'].iloc[0]:.2f}%")
cols[2].metric("Áp suất trung bình", f"{df['pressure'].mean():.2f} hPa",
f"{df['pressure'].iloc[-1] - df['pressure'].iloc[0]:.2f} hPa")
cols[3].metric("Cảm biến hoạt động", f"{df['sensor_id'].nunique()}")
# Tạo biểu đồ đường
with chart_placeholder.container():
# Biểu đồ nhiệt độ
temp_chart = alt.Chart(df).mark_line().encode(
x=alt.X('readable_time:T', title='Thời gian'),
y=alt.Y('temperature:Q', title='Nhiệt độ (°C)'),
color='sensor_id:N'
).properties(
title='Nhiệt độ theo thời gian thực',
height=300
)
# Biểu đồ độ ẩm
humidity_chart = alt.Chart(df).mark_line().encode(
x=alt.X('readable_time:T', title='Thời gian'),
y=alt.Y('humidity:Q', title='Độ ẩm (%)'),
color='sensor_id:N'
).properties(
title='Độ ẩm theo thời gian thực',
height=300
)
# Hiển thị biểu đồ
st.altair_chart(temp_chart, use_container_width=True)
st.altair_chart(humidity_chart, use_container_width=True)
# Hiển thị dữ liệu gần đây
with table_placeholder.container():
st.subheader("Dữ liệu gần nhất")
st.dataframe(df.tail(10)[['sensor_id', 'temperature', 'humidity', 'pressure', 'readable_time']])
# Đợi trước khi cập nhật
time.sleep(1)
except KeyboardInterrupt:
st.write("Đã dừng ứng dụng")
6.2 Chạy ứng dụng Streamlit
# Chạy ứng dụng
streamlit run app.py
7. Kiến trúc và triển khai hoàn chỉnh
7.1 Kiến trúc hoàn chỉnh
Một hệ thống phân tích dữ liệu thời gian thực hoàn chỉnh bao gồm các thành phần:
Nguồn dữ liệu:
- Cảm biến IoT
- API bên ngoài
- File logs
- Cơ sở dữ liệu
Kafka Cluster:
- Nhiều broker
- ZooKeeper (hoặc KRaft từ Kafka 3.x)
- Topics với nhiều partitions
Lớp xử lý dữ liệu:
- Consumers Python
- Apache Spark Streaming
- Faust Stream Processing
Lớp phân tích và học máy:
- Mô hình phát hiện bất thường
- Dự báo chuỗi thời gian
- Xử lý ngôn ngữ tự nhiên
Lớp trực quan hóa và cảnh báo:
- Dashboard Streamlit/Dash
- Hệ thống cảnh báo
- API REST với FastAPI/Flask
7.2 Code triển khai hoàn chỉnh
Tạo cấu trúc thư mục như sau:
real-time-analytics/
├── data_producers/
│ ├── sensor_producer.py
│ ├── api_producer.py
│ └── log_producer.py
├── data_consumers/
│ ├── raw_consumer.py
│ ├── analytics_consumer.py
│ └── ml_consumer.py
├── models/
│ ├── train_model.py
│ └── anomaly_detector.py
├── dashboard/
│ ├── app.py
│ ├── utils.py
│ └── assets/
├── config/
│ ├── kafka_config.py
│ └── app_config.py
├── utils/
│ ├── data_utils.py
│ └── kafka_utils.py
├── docker-compose.yml
├── requirements.txt
└── README.md
Ví dụ file config/kafka_config.py
"""
Cấu hình Kafka cho ứng dụng
"""
# Cấu hình chung
BOOTSTRAP_SERVERS = 'localhost:9092'
# Topics
SENSOR_TOPIC = 'sensor-data'
LOG_TOPIC = 'application-logs'
API_TOPIC = 'api-data'
ANALYZED_TOPIC = 'analyzed-data'
ANOMALY_TOPIC = 'anomaly-alerts'
# Consumer groups
RAW_CONSUMER_GROUP = 'raw-data-consumers'
ANALYTICS_CONSUMER_GROUP = 'analytics-consumers'
ML_CONSUMER_GROUP = 'ml-model-consumers'
DASHBOARD_CONSUMER_GROUP = 'dashboard-consumers'
# Producer config
producer_config = {
'bootstrap.servers': BOOTSTRAP_SERVERS,
'acks': 'all',
'retries': 3,
'retry.backoff.ms': 200,
'linger.ms': 5,
'batch.size': 16384,
}
# Consumer config
consumer_config = {
'bootstrap.servers': BOOTSTRAP_SERVERS,
'auto.offset.reset': 'latest',
'enable.auto.commit': True,
'auto.commit.interval.ms': 5000,
'max.poll.interval.ms': 300000,
'session.timeout.ms': 30000,
}
# Schema định nghĩa cho mỗi loại thông điệp
SENSOR_SCHEMA = {
'type': 'object',
'properties': {
'sensor_id': {'type': 'integer'},
'temperature': {'type': 'number'},
'humidity': {'type': 'number'},
'pressure': {'type': 'number'},
'timestamp': {'type': 'string', 'format': 'date-time'},
},
'required': ['sensor_id', 'temperature', 'timestamp']
}
ANOMALY_SCHEMA = {
'type': 'object',
'properties': {
'sensor_id': {'type': 'integer'},
'timestamp': {'type': 'string', 'format': 'date-time'},
'anomaly_type': {'type': 'string'},
'anomaly_score': {'type': 'number'},
'features': {'type': 'object'},
},
'required': ['sensor_id', 'timestamp', 'anomaly_type', 'anomaly_score']
}
Ví dụ file utils/kafka_utils.py
"""
Utilities for working with Kafka
"""
import json
from confluent_kafka import Producer, Consumer, KafkaError
from functools import wraps
import logging
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
)
logger = logging.getLogger(__name__)
def create_producer(config):
"""Create and return a Kafka producer with the specified config."""
producer = Producer(config)
logger.info("Created Kafka producer")
return producer
def create_consumer(config, topics):
"""Create and return a Kafka consumer subscribed to the specified topics."""
consumer = Consumer(config)
consumer.subscribe(topics)
logger.info(f"Created Kafka consumer subscribed to topics: {topics}")
return consumer
def send_message(producer, topic, key, value):
"""Send a message to the specified Kafka topic."""
try:
if isinstance(value, dict):
value = json.dumps(value).encode('utf-8')
elif not isinstance(value, bytes):
value = str(value).encode('utf-8')
if key and not isinstance(key, bytes):
key = str(key).encode('utf-8')
producer.produce(topic, key=key, value=value)
producer.poll(0) # Trigger delivery reports
return True
except Exception as e:
logger.error(f"Error sending message to topic {topic}: {e}")
return False
def flush_producer(producer):
"""Flush the producer to ensure all messages are sent."""
producer.flush()
logger.info("Producer flushed")
def consume_messages(consumer, batch_size=1, timeout=1.0):
"""
Consume messages from Kafka in batches.
Returns a list of messages or empty list if no messages are available.
"""
messages = []
try:
for _ in range(batch_size):
msg = consumer.poll(timeout)
if msg is None:
break
if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
# End of partition event
logger.info(f"Reached end of topic {msg.topic()} "
f"partition {msg.partition()}")
else:
# Error
logger.error(f"Consumer error: {msg.error()}")
break
# Extract and parse message
try:
value = msg.value().decode('utf-8')
data = json.loads(value)
messages.append(data)
except Exception as e:
logger.error(f"Error parsing message: {e}")
except Exception as e:
logger.error(f"Error consuming messages: {e}")
return messages
def close_consumer(consumer):
"""Close the Kafka consumer."""
consumer.close()
logger.info("Consumer closed")
def retry(max_attempts=3, backoff_factor=2):
"""
Retry decorator for Kafka operations.
Retries the function call with exponential backoff.
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
import time
attempts = 0
while attempts < max_attempts:
try:
return func(*args, **kwargs)
except Exception as e:
attempts += 1
if attempts == max_attempts:
logger.error(f"Failed after {max_attempts} attempts: {e}")
raise
wait_time = backoff_factor ** attempts
logger.warning(f"Retrying in {wait_time} seconds... (attempt {attempts}/{max_attempts})")
time.sleep(wait_time)
return wrapper
return decorator
Ví dụ file data_producers/sensor_producer.py
"""
Producer that simulates sensor data and sends it to Kafka
"""
import sys
import os
import time
import random
import json
from datetime import datetime
import threading
import logging
# Add parent directory to path to import from other modules
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from config.kafka_config import producer_config, SENSOR_TOPIC
from utils.kafka_utils import create_producer, send_message, flush_producer
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
)
logger = logging.getLogger(__name__)
class SensorDataProducer:
"""Simulates sensors sending temperature, humidity, and pressure data"""
def __init__(self, num_sensors=5, interval=1.0):
self.producer = create_producer(producer_config)
self.num_sensors = num_sensors
self.interval = interval
self.running = False
self.thread = None
# Base values for each sensor (to simulate different environments)
self.sensor_bases = {
i: {
'temp_base': random.uniform(18, 25),
'humid_base': random.uniform(40, 60),
'pressure_base': random.uniform(995, 1015)
}
for i in range(1, num_sensors + 1)
}
def generate_sensor_data(self, sensor_id):
"""Generate simulated sensor data with realistic variations"""
base = self.sensor_bases[sensor_id]
# Add some randomness to base values
temp = base['temp_base'] + random.uniform(-1.5, 1.5)
humid = base['humid_base'] + random.uniform(-5, 5)
pressure = base['pressure_base'] + random.uniform(-1, 1)
# Create data point
return {
'sensor_id': sensor_id,
'temperature': round(temp, 2),
'humidity': round(humid, 2),
'pressure': round(pressure, 2),
'timestamp': datetime.now().isoformat()
}
def produce_data(self):
"""Main loop for producing sensor data"""
while self.running:
for sensor_id in range(1, self.num_sensors + 1):
# Generate data for this sensor
data = self.generate_sensor_data(sensor_id)
# Send to Kafka
success = send_message(
self.producer,
SENSOR_TOPIC,
str(sensor_id),
data
)
if success:
logger.info(f"Sent data from sensor {sensor_id}: temp={data['temperature']}°C, "
f"humid={data['humidity']}%, press={data['pressure']} hPa")
# Sleep until next interval
time.sleep(self.interval)
def start(self):
"""Start the producer in a separate thread"""
if not self.running:
self.running = True
self.thread = threading.Thread(target=self.produce_data)
self.thread.daemon = True
self.thread.start()
logger.info(f"Started sensor data producer with {self.num_sensors} sensors")
def stop(self):
"""Stop the producer"""
if self.running:
self.running = False
if self.thread:
self.thread.join(timeout=2.0)
flush_producer(self.producer)
logger.info("Stopped sensor data producer")
def __enter__(self):
self.start()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.stop()
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description='Sensor data producer for Kafka')
parser.add_argument('--sensors', type=int, default=5, help='Number of sensors to simulate')
parser.add_argument('--interval', type=float, default=1.0, help='Interval between data points (seconds)')
args = parser.parse_args()
try:
with SensorDataProducer(num_sensors=args.sensors, interval=args.interval) as producer:
# Keep main thread alive
while True:
time.sleep(1)
except KeyboardInterrupt:
logger.info("Producer interrupted by user")
Ví dụ file models/anomaly_detector.py
"""
Anomaly detection model for sensor data
"""
import os
import pickle
import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import logging
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
)
logger = logging.getLogger(__name__)
class AnomalyDetector:
"""Anomaly detection model using Isolation Forest algorithm"""
def __init__(self, model_dir='../models/saved'):
self.model_dir = model_dir
self.model_path = os.path.join(model_dir, 'isolation_forest.pkl')
self.scaler_path = os.path.join(model_dir, 'scaler.pkl')
# Create model directory if it doesn't exist
os.makedirs(model_dir, exist_ok=True)
# Initialize model and scaler
self.model = None
self.scaler = StandardScaler()
# Feature columns used for anomaly detection
self.feature_columns = ['temperature', 'humidity', 'pressure']
# Try to load existing model
self.load_model()
def load_model(self):
"""Load the model and scaler if they exist"""
try:
if os.path.exists(self.model_path) and os.path.exists(self.scaler_path):
with open(self.model_path, 'rb') as f:
self.model = pickle.load(f)
with open(self.scaler_path, 'rb') as f:
self.scaler = pickle.load(f)
logger.info("Loaded existing anomaly detection model and scaler")
return True
else:
logger.info("No existing model found")
return False
except Exception as e:
logger.error(f"Error loading model: {e}")
return False
def save_model(self):
"""Save the model and scaler to disk"""
try:
with open(self.model_path, 'wb') as f:
pickle.dump(self.model, f)
with open(self.scaler_path, 'wb') as f:
pickle.dump(self.scaler, f)
logger.info("Saved anomaly detection model and scaler")
return True
except Exception as e:
logger.error(f"Error saving model: {e}")
return False
def train(self, data, contamination=0.05):
"""
Train the anomaly detection model
Args:
data (pd.DataFrame): DataFrame containing sensor data
contamination (float): Proportion of outliers in the data
Returns:
bool: True if training was successful
"""
try:
# Extract features
X = data[self.feature_columns].values
# Fit scaler
self.scaler.fit(X)
X_scaled = self.scaler.transform(X)
# Train model
self.model = IsolationForest(
n_estimators=100,
max_samples='auto',
contamination=contamination,
random_state=42
)
self.model.fit(X_scaled)
# Save model
self.save_model()
logger.info(f"Trained anomaly detection model on {len(data)} samples")
return True
except Exception as e:
logger.error(f"Error training model: {e}")
return False
def predict(self, data):
"""
Predict anomalies in the data
Args:
data (pd.DataFrame): DataFrame containing sensor data
Returns:
pd.DataFrame: Original data with anomaly predictions and scores
"""
if self.model is None:
logger.error("No model has been trained yet")
return None
try:
# Extract features
X = data[self.feature_columns].values
# Scale features
X_scaled = self.scaler.transform(X)
# Predict
# -1 for anomalies, 1 for normal
predictions = self.model.predict(X_scaled)
# Get anomaly scores
scores = self.model.decision_function(X_scaled)
# Add predictions and scores to data
result = data.copy()
result['anomaly'] = predictions
result['anomaly_score'] = scores
# Mark anomalies (predictions of -1)
result['is_anomaly'] = result['anomaly'] == -1
return result
except Exception as e:
logger.error(f"Error making predictions: {e}")
return None
def identify_anomaly_type(self, row):
"""
Identify the type of anomaly based on feature values
Args:
row (pd.Series): Row of data with predictions
Returns:
str: Description of anomaly type
"""
if row['anomaly'] != -1:
return "normal"
# Check which feature is most abnormal
temp_z = abs((row['temperature'] - self.scaler.mean_[0]) / self.scaler.scale_[0])
humid_z = abs((row['humidity'] - self.scaler.mean_[1]) / self.scaler.scale_[1])
pressure_z = abs((row['pressure'] - self.scaler.mean_[2]) / self.scaler.scale_[2])
max_z = max(temp_z, humid_z, pressure_z)
if temp_z == max_z:
return "temperature_anomaly"
elif humid_z == max_z:
return "humidity_anomaly"
else:
return "pressure_anomaly"
Ví dụ file dashboard/app.py (sử dụng FastAPI và JavaScript WebSocket)
"""
Dashboard for real-time sensor data visualization with FastAPI and WebSockets
"""
import sys
import os
import asyncio
import json
import logging
from datetime import datetime
from typing import List, Dict, Any
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
import uvicorn
# Add parent directory to path
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from config.kafka_config import consumer_config, SENSOR_TOPIC, ANOMALY_TOPIC
from utils.kafka_utils import create_consumer, consume_messages, close_consumer
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
)
logger = logging.getLogger(__name__)
app = FastAPI(title="Real-time Sensor Analytics Dashboard")
# Create WebSocket connection manager
class ConnectionManager:
def __init__(self):
self.active_connections: List[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
# Create Kafka consumer
sensor_consumer_config = consumer_config.copy()
sensor_consumer_config['group.id'] = 'dashboard-sensor-group'
sensor_consumer = create_consumer(sensor_consumer_config, [SENSOR_TOPIC])
anomaly_consumer_config = consumer_config.copy()
anomaly_consumer_config['group.id'] = 'dashboard-anomaly-group'
anomaly_consumer = create_consumer(anomaly_consumer_config, [ANOMALY_TOPIC])
# Store recent data
recent_sensor_data = []
max_data_points = 500
recent_anomalies = []
max_anomalies = 50
# HTML response
@app.get("/", response_class=HTMLResponse)
async def get():
html_content = """
<!DOCTYPE html>
<html>
<head>
<title>Real-time Sensor Analytics Dashboard</title>
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/css/bootstrap.min.css" rel="stylesheet">
<script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
<style>
body { padding: 20px; }
.sensor-card { margin-bottom: 20px; }
.chart-container { position: relative; height: 250px; }
.anomaly-alert { color: white; background-color: #dc3545; border-radius: 5px; padding: 10px; margin-bottom: 10px; }
</style>
</head>
<body>
<div class="container">
<h1 class="mt-4 mb-4">Real-time Sensor Analytics Dashboard</h1>
<div class="row">
<div class="col-md-8">
<div class="card">
<div class="card-header">Temperature (°C)</div>
<div class="card-body">
<div class="chart-container">
<canvas id="tempChart"></canvas>
</div>
</div>
</div>
</div>
<div class="col-md-4">
<div class="card">
<div class="card-header">Latest Metrics</div>
<div class="card-body" id="metrics">
<p>Waiting for data...</p>
</div>
</div>
</div>
</div>
<div class="row mt-4">
<div class="col-md-6">
<div class="card">
<div class="card-header">Humidity (%)</div>
<div class="card-body">
<div class="chart-container">
<canvas id="humidChart"></canvas>
</div>
</div>
</div>
</div>
<div class="col-md-6">
<div class="card">
<div class="card-header">Pressure (hPa)</div>
<div class="card-body">
<div class="chart-container">
<canvas id="pressChart"></canvas>
</div>
</div>
</div>
</div>
</div>
<div class="row mt-4">
<div class="col-md-12">
<div class="card">
<div class="card-header">
Anomaly Alerts
<span id="anomaly-count" class="badge bg-danger ms-2">0</span>
</div>
<div class="card-body">
<div id="anomalies">No anomalies detected</div>
</div>
</div>
</div>
</div>
<div class="row mt-4">
<div class="col-md-12">
<div class="card">
<div class="card-header">Latest Data</div>
<div class="card-body">
<div style="height: 200px; overflow-y: auto;">
<table class="table table-striped">
<thead>
<tr>
<th>Time</th>
<th>Sensor</th>
<th>Temperature</th>
<th>Humidity</th>
<th>Pressure</th>
</tr>
</thead>
<tbody id="dataTable">
</tbody>
</table>
</div>
</div>
</div>
</div>
</div>
</div>
<script>
// Initialize charts
const colors = [
'rgba(255, 99, 132, 1)',
'rgba(54, 162, 235, 1)',
'rgba(255, 206, 86, 1)',
'rgba(75, 192, 192, 1)',
'rgba(153, 102, 255, 1)',
'rgba(255, 159, 64, 1)'
];
// Temperature chart
const tempCtx = document.getElementById('tempChart').getContext('2d');
const tempChart = new Chart(tempCtx, {
type: 'line',
data: {
datasets: []
},
options: {
responsive: true,
maintainAspectRatio: false,
scales: {
x: {
type: 'time',
time: {
unit: 'minute'
}
},
y: {
title: {
display: true,
text: 'Temperature (°C)'
}
}
}
}
});
// Humidity chart
const humidCtx = document.getElementById('humidChart').getContext('2d');
const humidChart = new Chart(humidCtx, {
type: 'line',
data: {
datasets: []
},
options: {
responsive: true,
maintainAspectRatio: false,
scales: {
x: {
type: 'time',
time: {
unit: 'minute'
}
},
y: {
title: {
display: true,
text: 'Humidity (%)'
}
}
}
}
});
// Pressure chart
const pressCtx = document.getElementById('pressChart').getContext('2d');
const pressChart = new Chart(pressCtx, {
type: 'line',
data: {
datasets: []
},
options: {
responsive: true,
maintainAspectRatio: false,
scales: {
x: {
type: 'time',
time: {
unit: 'minute'
}
},
y: {
title: {
display: true,
text: 'Pressure (hPa)'
}
}
}
}
});
// Sensor data by ID
const sensorData = {};
let anomalyCount = 0;
// Connect to WebSocket
const ws = new WebSocket(`ws://${window.location.host}/ws`);
ws.onmessage = function(event) {
const data = JSON.parse(event.data);
if (data.type === 'sensor') {
processSensorData(data);
} else if (data.type === 'anomaly') {
processAnomalyData(data);
}
};
function processSensorData(data) {
const sensorId = data.sensor_id;
const timestamp = new Date(data.timestamp);
// Initialize sensor data if not exists
if (!sensorData[sensorId]) {
sensorData[sensorId] = {
temperature: [],
humidity: [],
pressure: []
};
// Add dataset to charts
tempChart.data.datasets.push({
label: `Sensor ${sensorId}`,
data: [],
borderColor: colors[sensorId % colors.length],
fill: false,
tension: 0.1
});
humidChart.data.datasets.push({
label: `Sensor ${sensorId}`,
data: [],
borderColor: colors[sensorId % colors.length],
fill: false,
tension: 0.1
});
pressChart.data.datasets.push({
label: `Sensor ${sensorId}`,
data: [],
borderColor: colors[sensorId % colors.length],
fill: false,
tension: 0.1
});
}
// Add data point to dataset
sensorData[sensorId].temperature.push({x: timestamp, y: data.temperature});
sensorData[sensorId].humidity.push({x: timestamp, y: data.humidity});
sensorData[sensorId].pressure.push({x: timestamp, y: data.pressure});
// Limit dataset size
const maxPoints = 100;
if (sensorData[sensorId].temperature.length > maxPoints) {
sensorData[sensorId].temperature.shift();
sensorData[sensorId].humidity.shift();
sensorData[sensorId].pressure.shift();
}
// Update charts
tempChart.data.datasets[sensorId-1].data = sensorData[sensorId].temperature;
humidChart.data.datasets[sensorId-1].data = sensorData[sensorId].humidity;
pressChart.data.datasets[sensorId-1].data = sensorData[sensorId].pressure;
tempChart.update();
humidChart.update();
pressChart.update();
// Update metrics
updateMetrics();
// Add to data table
addDataRow(data);
}
function processAnomalyData(data) {
anomalyCount++;
document.getElementById('anomaly-count').textContent = anomalyCount;
// Add anomaly alert
const anomaliesDiv = document.getElementById('anomalies');
if (anomaliesDiv.textContent === 'No anomalies detected') {
anomaliesDiv.innerHTML = '';
}
// Create alert element
const alert = document.createElement('div');
alert.className = 'anomaly-alert';
alert.innerHTML = `
<strong>Anomaly detected!</strong><br>
Sensor: ${data.sensor_id}<br>
Time: ${new Date(data.timestamp).toLocaleString()}<br>
Type: ${data.anomaly_type}<br>
Score: ${data.anomaly_score.toFixed(3)}
`;
// Add to alerts
anomaliesDiv.prepend(alert);
// Limit number of alerts
const maxAlerts = 5;
const alerts = anomaliesDiv.getElementsByClassName('anomaly-alert');
if (alerts.length > maxAlerts) {
anomaliesDiv.removeChild(alerts[alerts.length - 1]);
}
}
function updateMetrics() {
// Calculate latest metrics across all sensors
let totalTemp = 0;
let totalHumid = 0;
let totalPress = 0;
let count = 0;
for (const sensorId in sensorData) {
const sensorDataset = sensorData[sensorId];
if (sensorDataset.temperature.length > 0) {
totalTemp += sensorDataset.temperature[sensorDataset.temperature.length - 1].y;
totalHumid += sensorDataset.humidity[sensorDataset.humidity.length - 1].y;
totalPress += sensorDataset.pressure[sensorDataset.pressure.length - 1].y;
count++;
}
}
if (count > 0) {
const avgTemp = totalTemp / count;
const avgHumid = totalHumid / count;
const avgPress = totalPress / count;
// Update metrics display
document.getElementById('metrics').innerHTML = `
<div class="row">
<div class="col-6">
<h5>Temperature</h5>
<h3>${avgTemp.toFixed(1)}°C</h3>
</div>
<div class="col-6">
<h5>Humidity</h5>
<h3>${avgHumid.toFixed(1)}%</h3>
</div>
</div>
<div class="row mt-3">
<div class="col-6">
<h5>Pressure</h5>
<h3>${avgPress.toFixed(1)} hPa</h3>
</div>
<div class="col-6">
<h5>Active Sensors</h5>
<h3>${count}</h3>
</div>
</div>
`;
}
}
function addDataRow(data) {
const table = document.getElementById('dataTable');
const row = table.insertRow(0);
// Format timestamp
const timestamp = new Date(data.timestamp).toLocaleTimeString();
// Insert cells
const cell1 = row.insertCell(0);
const cell2 = row.insertCell(1);
const cell3 = row.insertCell(2);
const cell4 = row.insertCell(3);
const cell5 = row.insertCell(4);
cell1.innerHTML = timestamp;
cell2.innerHTML = data.sensor_id;
cell3.innerHTML = `${data.temperature.toFixed(1)}°C`;
cell4.innerHTML = `${data.humidity.toFixed(1)}%`;
cell5.innerHTML = `${data.pressure.toFixed(1)} hPa`;
// Limit table rows
const maxRows = 100;
if (table.rows.length > maxRows) {
table.deleteRow(table.rows.length - 1);
}
}
</script>
</body>
</html>
"""
return html_content
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
# This is just to keep the connection alive
# The actual data will be sent from the Kafka consumer task
await asyncio.sleep(1)
except WebSocketDisconnect:
manager.disconnect(websocket)
async def kafka_sensor_consumer_task():
"""Background task to consume sensor data from Kafka and broadcast to WebSocket clients"""
while True:
try:
# Consume messages
messages = consume_messages(sensor_consumer, batch_size=5, timeout=0.1)
# Add timestamp for frontend
for message in messages:
if 'timestamp' in message:
# Parse ISO format timestamp
message['timestamp'] = datetime.fromisoformat(message['timestamp']).isoformat()
else:
message['timestamp'] = datetime.now().isoformat()
# Add message type
message['type'] = 'sensor'
# Store recent data
recent_sensor_data.append(message)
if len(recent_sensor_data) > max_data_points:
recent_sensor_data.pop(0)
# Broadcast to WebSocket clients
await manager.broadcast(json.dumps(message))
except Exception as e:
logger.error(f"Error in sensor consumer task: {e}")
# Brief pause
await asyncio.sleep(0.1)
async def kafka_anomaly_consumer_task():
"""Background task to consume anomaly data from Kafka and broadcast to WebSocket clients"""
while True:
try:
# Consume messages
messages = consume_messages(anomaly_consumer, batch_size=5, timeout=0.1)
# Process and broadcast anomalies
for message in messages:
# Add message type
message['type'] = 'anomaly'
# Store recent anomalies
recent_anomalies.append(message)
if len(recent_anomalies) > max_anomalies:
recent_anomalies.pop(0)
# Broadcast to WebSocket clients
await manager.broadcast(json.dumps(message))
except Exception as e:
logger.error(f"Error in anomaly consumer task: {e}")
# Brief pause
await asyncio.sleep(0.1)
@app.on_event("startup")
async def startup_event():
# Start Kafka consumer tasks
asyncio.create_task(kafka_sensor_consumer_task())
asyncio.create_task(kafka_anomaly_consumer_task())
logger.info("Started Kafka consumer tasks")
@app.on_event("shutdown")
def shutdown_event():
# Close Kafka consumers
close_consumer(sensor_consumer)
close_consumer(anomaly_consumer)
logger.info("Closed Kafka consumers")
# API endpoints for recent data
@app.get("/api/sensor-data")
def get_sensor_data():
return recent_sensor_data
@app.get("/api/anomalies")
def get_anomalies():
return recent_anomalies
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
8. Các phương pháp tối ưu hiệu suất
8.1 Tối ưu hóa Producer
- Sử dụng batching và compression
- Tăng cường độ tin cậy với acks
- Lập lịch và retry logic
- Caching và buffering
# Cấu hình Producer được tối ưu
optimized_producer_config = {
'bootstrap.servers': 'localhost:9092',
'acks': 'all', # Đảm bảo độ tin cậy cao nhất
'retries': 3, # Số lần thử lại
'retry.backoff.ms': 200, # Thời gian chờ giữa các lần thử lại
'linger.ms': 5, # Thời gian chờ để batch tin nhắn
'batch.size': 16384, # Kích thước batch
'compression.type': 'snappy', # Nén dữ liệu
'buffer.memory': 33554432, # Bộ nhớ đệm (32MB)
'key.serializer': 'StringSerializer',
'value.serializer': 'StringSerializer'
}
8.2 Tối ưu hóa Consumer
- Sử dụng consumer groups và partition rebalancing
- Xử lý messages theo batch
- Parallel processing
- Checkpoint và commit strategy
# Cấu hình Consumer được tối ưu
optimized_consumer_config = {
'bootstrap.servers': 'localhost:9092',
'group.id': 'optimized-consumer-group',
'auto.offset.reset': 'earliest',
'enable.auto.commit': False, # Kiểm soát commit thủ công
'max.poll.records': 500, # Số lượng tin nhắn tối đa mỗi lần poll
'max.poll.interval.ms': 300000, # Thời gian tối đa giữa các lần poll
'session.timeout.ms': 30000, # Thời gian timeout phiên
'heartbeat.interval.ms': 10000, # Thời gian gửi heartbeat
'fetch.min.bytes': 1024, # Số byte tối thiểu để fetch
'fetch.max.bytes': 52428800, # Số byte tối đa để fetch (50MB)
'fetch.max.wait.ms': 500 # Thời gian chờ tối đa để fetch
}
8.3 Xử lý song song
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
import os
def process_data_parallel(data_batch):
"""Process a batch of data using parallel processing"""
# Number of worker processes/threads
num_workers = os.cpu_count()
# Split data into chunks for parallel processing
chunk_size = max(1, len(data_batch) // num_workers)
chunks = [data_batch[i:i + chunk_size] for i in range(0, len(data_batch), chunk_size)]
results = []
# Use ProcessPoolExecutor for CPU-bound tasks
with ProcessPoolExecutor(max_workers=num_workers) as executor:
# Map each chunk to a worker
chunk_results = executor.map(process_chunk, chunks)
# Collect results
for result in chunk_results:
results.extend(result)
return results
def process_chunk(data_chunk):
"""Process a chunk of data in a worker process"""
results = []
for item in data_chunk:
# Perform CPU-intensive processing
processed_item = complex_processing(item)
results.append(processed_item)
return results
def complex_processing(item):
"""Example of a complex processing function"""
# Simulate complex processing
result = item.copy()
# Feature engineering
result['temp_squared'] = result['temperature'] ** 2
result['humid_normalized'] = result['humidity'] / 100
# Calculate moving averages
# (in practice, this would use data from multiple items)
result['temp_trend'] = 0
return result
8.4 Lưu trữ dữ liệu hiệu quả
import pandas as pd
import numpy as np
import h5py
import os
from datetime import datetime, timedelta
class TimeSeriesStorage:
"""Efficient storage for time series data"""
def __init__(self, file_path, max_records=1000000):
self.file_path = file_path
self.max_records = max_records
self.buffer = []
self.buffer_size = 10000
# Create file if it doesn't exist
if not os.path.exists(file_path):
self._initialize_file()
def _initialize_file(self):
"""Initialize HDF5 file structure"""
with h5py.File(self.file_path, 'w') as f:
# Create groups for different time scales
f.create_group('hourly')
f.create_group('daily')
f.create_group('monthly')
# Create timestamp datasets
f.create_dataset('hourly/timestamps', (0,), maxshape=(None,), dtype='i8')
f.create_dataset('daily/timestamps', (0,), maxshape=(None,), dtype='i8')
f.create_dataset('monthly/timestamps', (0,), maxshape=(None,), dtype='i8')
# Create data datasets for each sensor
for i in range(1, 11): # Assuming 10 sensors
f.create_dataset(f'hourly/sensor_{i}/temperature', (0,), maxshape=(None,), dtype='f4')
f.create_dataset(f'hourly/sensor_{i}/humidity', (0,), maxshape=(None,), dtype='f4')
f.create_dataset(f'hourly/sensor_{i}/pressure', (0,), maxshape=(None,), dtype='f4')
f.create_dataset(f'daily/sensor_{i}/temperature', (0,), maxshape=(None,), dtype='f4')
f.create_dataset(f'daily/sensor_{i}/humidity', (0,), maxshape=(None,), dtype='f4')
f.create_dataset(f'daily/sensor_{i}/pressure', (0,), maxshape=(None,), dtype='f4')
f.create_dataset(f'monthly/sensor_{i}/temperature', (0,), maxshape=(None,), dtype='f4')
f.create_dataset(f'monthly/sensor_{i}/humidity', (0,), maxshape=(None,), dtype='f4')
f.create_dataset(f'monthly/sensor_{i}/pressure', (0,), maxshape=(None,), dtype='f4')
def add_record(self, record):
"""Add a single record to the buffer"""
self.buffer.append(record)
# Flush buffer when it reaches buffer_size
if len(self.buffer) >= self.buffer_size:
self.flush()
def flush(self):
"""Write buffered records to storage"""
if not self.buffer:
return
# Convert buffer to DataFrame for easier processing
df = pd.DataFrame(self.buffer)
# Reset buffer
self.buffer = []
# Process data for hourly storage
self._update_hourly_data(df)
# Update aggregated daily and monthly data if needed
self._update_aggregated_data()
def _update_hourly_data(self, df):
"""Update hourly data storage"""
# Convert timestamps to Unix timestamps (seconds since epoch)
df['timestamp_unix'] = pd.to_datetime(df['timestamp']).astype(np.int64) // 10**9
with h5py.File(self.file_path, 'a') as f:
# Get current hourly timestamps
timestamps = f['hourly/timestamps'][:]
# Add new timestamps
new_timestamps = np.unique(df['timestamp_unix'].values)
all_timestamps = np.unique(np.concatenate([timestamps, new_timestamps]))
# Resize timestamps dataset
f['hourly/timestamps'].resize((len(all_timestamps),))
f['hourly/timestamps'][:] = all_timestamps
# Update data for each sensor
for sensor_id in df['sensor_id'].unique():
sensor_df = df[df['sensor_id'] == sensor_id]
# Ensure sensor datasets exist
if f'hourly/sensor_{sensor_id}/temperature' not in f:
f.create_dataset(f'hourly/sensor_{sensor_id}/temperature', (len(all_timestamps),), dtype='f4')
f.create_dataset(f'hourly/sensor_{sensor_id}/humidity', (len(all_timestamps),), dtype='f4')
f.create_dataset(f'hourly/sensor_{sensor_id}/pressure', (len(all_timestamps),), dtype='f4')
else:
# Resize datasets if needed
f[f'hourly/sensor_{sensor_id}/temperature'].resize((len(all_timestamps),))
f[f'hourly/sensor_{sensor_id}/humidity'].resize((len(all_timestamps),))
f[f'hourly/sensor_{sensor_id}/pressure'].resize((len(all_timestamps),))
# Update data
for _, row in sensor_df.iterrows():
idx = np.where(all_timestamps == row['timestamp_unix'])[0][0]
f[f'hourly/sensor_{sensor_id}/temperature'][idx] = row['temperature']
f[f'hourly/sensor_{sensor_id}/humidity'][idx] = row['humidity']
f[f'hourly/sensor_{sensor_id}/pressure'][idx] = row['pressure']
def _update_aggregated_data(self):
"""Update daily and monthly aggregated data"""
# This would aggregate hourly data into daily and monthly summaries
# Implementation depends on specific aggregation requirements
pass
def get_data(self, sensor_id, start_time, end_time, resolution='hourly'):
"""Retrieve data for a specific sensor and time range"""
# Convert time range to Unix timestamps
start_unix = int(pd.to_datetime(start_time).timestamp())
end_unix = int(pd.to_datetime(end_time).timestamp())
with h5py.File(self.file_path, 'r') as f:
# Get timestamps
timestamps = f[f'{resolution}/timestamps'][:]
# Filter timestamps in range
mask = (timestamps >= start_unix) & (timestamps <= end_unix)
filtered_timestamps = timestamps[mask]
# Get data
temperature = f[f'{resolution}/sensor_{sensor_id}/temperature'][mask]
humidity = f[f'{resolution}/sensor_{sensor_id}/humidity'][mask]
pressure = f[f'{resolution}/sensor_{sensor_id}/pressure'][mask]
# Convert timestamps to datetime
dates = [datetime.fromtimestamp(ts) for ts in filtered_timestamps]
# Create DataFrame
df = pd.DataFrame({
'timestamp': dates,
'temperature': temperature,
'humidity': humidity,
'pressure': pressure
})
return df
9. Tình huống thực tế và các trường hợp sử dụng
9.1 Giám sát hệ thống CNTT
- Theo dõi logs và metrics theo thời gian thực
- Phát hiện sự cố sớm
- Tự động hóa phản ứng với sự cố
9.2 Giám sát và điều khiển nhà máy
- Theo dõi các cảm biến IoT
- Tối ưu hóa quy trình sản xuất
- Bảo trì dự đoán (Predictive maintenance)
9.3 Phân tích thị trường tài chính
- Phát hiện cơ hội giao dịch
- Quản lý rủi ro theo thời gian thực
- Phân tích tâm lý thị trường
9.4 Theo dõi người dùng và trải nghiệm người dùng
- Phân tích hành vi người dùng
- A/B testing theo thời gian thực
- Cá nhân hóa nội dung động
Kết luận
Kết hợp Apache Kafka và Python tạo nên một giải pháp mạnh mẽ cho việc xây dựng hệ thống phân tích dữ liệu thời gian thực. Kafka cung cấp khả năng xử lý luồng dữ liệu phân tán, độ tin cậy cao và khả năng mở rộng, trong khi Python mang đến sự linh hoạt, năng suất và hệ sinh thái phong phú các thư viện phân tích dữ liệu và học máy.
Bằng cách tuân theo các thực hành tốt, tối ưu hóa hiệu suất và thiết kế kiến trúc phù hợp, bạn có thể xây dựng các ứng dụng phân tích thời gian thực đáp ứng các yêu cầu khắt khe trong môi trường doanh nghiệp.
Việc nắm vững những công nghệ này mở ra cơ hội lớn để biến đổi dữ liệu thô thành thông tin có giá trị, từ đó đưa ra quyết định kinh doanh nhanh chóng và hiệu quả.