Stream processing thời gian thực và StreamSensing
Batch và stream, cửa sổ trượt, trễ và thông lượng, framework StreamSensing xử lý multimedia thời gian thực.
KhóThuật toán ~60 phút
Ôn lại trước khi học
Giới thiệu
Bài này học xử lý dòng sự kiện ngay khi tới. Dùng để bắt gian lận thẻ và cảnh báo camera.
Kiến thức cần nhớ
Batch và stream
- Định nghĩa: batch gom mẻ rồi tính, stream tính ngay.
- Batch trễ phút giờ, hợp báo cáo cuối ngày.
- Stream trễ mili giây, hợp giá cổ phiếu và cảm biến.
- Ví dụ đời thường: như phát trực tiếp thay vì chờ cuối ngày.
- Stream phải chịu dữ liệu muộn và rối thứ tự.
Cửa sổ và watermark
- Định nghĩa: cửa sổ cắt dòng vô hạn thành miếng tính được.
- Trượt: tính N phút gần nhất, cập nhật liên tục.
- Cố định (tumbling): chia thành mẻ rời nhau, mỗi sự kiện thuộc đúng 1 cửa sổ, tiết kiệm hơn.
- Ví dụ đời thường: như chỉ nhìn 5 phút camera gần nhất.
- Watermark là mốc an toàn, sự kiện sau mốc là muộn.
| Kiểu gửi | Đặc điểm | Khi dùng |
|---|---|---|
| Ít nhất 1 lần | Dễ trùng, cần khử | Không được mất |
| Tối đa 1 lần | Nhanh, có thể mất | Chấp nhận sót |
| Đúng 1 lần | Chuẩn, tốn kém | Tiền tệ, thanh toán |
StreamSensing
- Định nghĩa: framework trích đặc trưng ngay trên dòng video.
- Chỉ lưu đoạn đáng ngờ thay vì cả video dài.
- Ví dụ đời thường: như bảo vệ chỉ ghi lúc có động.
- Giảm trễ từ phút xuống giây nhờ cửa sổ.
- Khi quá tải thì giữ sự kiện quan trọng trước.
Ví dụ
- Input: giao dịch thẻ lúc 10 giờ -> Output: cảnh báo trong 1 giây.
- Giải thích: stream tính ngay thay vì chờ batch đêm.
# Trung bình trượt trên cửa sổ gần nhất
cua_so = giu_n_phut_cuoi(su_kien, n=5) # cắt dòng vô hạn
tb = sum(cua_so) / len(cua_so) # tính ngay
print(tb)
💻 Bài tập
Ngôn ngữ: Python 3 (thư viện chuẩn và asyncio). Không dùng framework stream có sẵn cho Bài 1-2. Nộp vào day-105/.
Bài 1: Cửa sổ thời gian
Mỗi sự kiện: Event(ts: float, key: str, value: float), với ts là thời gian sự kiện (giây).
| Lớp | Yêu cầu |
|---|---|
TumblingWindow(size) | Cửa sổ rời [0,10), [10,20)... Đóng cửa sổ thì phát (start, key, count, sum, avg) |
SlidingWindow(size, slide) | Cửa sổ dài size, nhảy mỗi slide giây. 1 sự kiện thuộc nhiều cửa sổ |
SessionWindow(gap) | Nhóm sự kiện cùng key cách nhau dưới gap giây |
Tumbling(10): sự kiện ts = 1, 3, 12, 15, 25 -> [0,10): 2 sự kiện, [10,20): 2, [20,30): 1
Sliding(10, 5): ts = 7 thuộc [0,10) và [5,15)
Session(gap=5): ts = 1, 3, 20, 22 -> 2 phiên
Bài 2: Watermark và dữ liệu muộn
- Sự kiện đến không theo thứ tự, trễ ngẫu nhiên 0-8 giây.
- Watermark =
max_ts_đã_thấy - 5. Cửa sổ chỉ đóng khi watermark vượt cuối cửa sổ. - Sự kiện đến sau khi cửa sổ đã đóng: đưa vào
late_eventsvà đếm. - Thử độ trễ cho phép 0, 5, 10 giây. In số sự kiện muộn bị bỏ và độ trễ phát kết quả trung bình.
Bài 3: Phát hiện gian lận thẻ thời gian thực (nâng cao)
Luồng giao dịch (ts, card_id, amount, city) sinh bằng asyncio, 1000 sự kiện/giây.
- Cảnh báo khi 1 thẻ có hơn 5 giao dịch trong 60 giây (cửa sổ trượt).
- Cảnh báo khi 1 thẻ giao dịch ở 2 thành phố cách nhau dưới 10 phút.
- Đo độ trễ từ lúc sự kiện đến tới lúc phát cảnh báo (p50, p99). Mục tiêu dưới 1 giây.
- Quá tải: tăng lên 10.000 sự kiện/giây với hàng đợi giới hạn. Khi đầy, bỏ giao dịch nhỏ hơn 100.000đ trước. Đếm số bị bỏ.
✅ Tiêu chí hoàn thành
- Test mẫu 3 loại cửa sổ pass.
- Bảng đánh đổi độ trễ cho phép và số sự kiện muộn.
- Bài 3 phát cảnh báo đúng với dữ liệu cài sẵn gian lận, có số p50/p99.
- Giải thích được khác biệt giữa thời gian sự kiện và thời gian xử lý.
❓ Câu hỏi ôn tập
- Batch khác stream ở đâu?
- Cửa sổ trượt là gì?
- Watermark để làm gì?
- Khi nào cần đúng 1 lần?
🌕 Bạn đã hoàn thành ngày 105. Làm hết bài tập phía trên rồi sang ngày tiếp theo.