Một cách tiếp cận đơn giản để hiểu và ứng dụng kiến trúc event-driven trong bài toán thị giác máy tính.
Lời mở đầu
Mình luôn tò mò về cách các hệ thống thời gian thực hoạt động — đặc biệt là khi theo dõi đối tượng bằng thị giác máy tính. Gần đây, mình có xem một video của anh Mì AI về triển khai AI với Kafka. Vậy nên mình đã quyết định làm một dự án nhỏ thú vị: kết hợp Kafka, một hệ thống truyền tin mạnh mẽ, với YOLO, một trong những mô hình phát hiện đối tượng nhanh nhất hiện nay.
Trong bài viết này, mình sẽ hướng dẫn bạn từng bước để thiết lập một pipeline đơn giản: gửi khung hình video qua Kafka, xử lý chúng bằng YOLO, và hiển thị kết quả theo thời gian thực. Bạn không cần hệ thống quá phức tạp — chỉ cần một vài công cụ cơ bản, một chút Python, và tinh thần khám phá.
Nếu bạn thấy hay, đừng quên thả like hoặc chia sẻ cho người bạn cũng đang muốn thử điều gì đó mới mẻ nhé!
Bài viết gốc của mình tại đây.
Kafka — Nền tảng truyền sự kiện
Trước khi đi vào phần theo dõi, hãy nói một chút về nền tảng cốt lõi của dự án này — Kafka.
Kafka là một nền tảng truyền sự kiện phân tán. Nói đơn giản, nó giống như một “xe buýt” truyền tin tốc độ cao cho phép các thành phần trong hệ thống của bạn giao tiếp với nhau theo thời gian thực. Thay vì đợi một bước hoàn tất rồi mới tới bước tiếp theo, Kafka giúp dữ liệu luân chuyển liên tục giữa các phần.
Trong dự án này, mình sử dụng Kafka để truyền các khung hình từ một producer (có thể là camera hoặc đoạn script đọc video) tới một consumer (YOLO). Cách tiếp cận này giúp dễ mở rộng, dễ giám sát và quản lý từng phần riêng biệt. Quan trọng hơn, cảm giác giống như bạn đang xây dựng một hệ thống thật sự — không chỉ là chạy YOLO trong một vòng lặp.
Điểm mình thích ở Kafka là: nhanh, đáng tin cậy và có khả năng mở rộng. Khi bạn hiểu nguyên lý hoạt động, bạn có thể áp dụng nó cho nhiều bài toán khác nhau — từ tổng hợp log đến kiến trúc microservices theo hướng sự kiện.
YOLO — Phát hiện đối tượng thời gian thực
Khi Kafka đã xử lý phần stream video, giờ là lúc nói về mảnh ghép thứ hai: YOLOv8.
YOLO (You Only Look Once) là một mô hình phát hiện đối tượng thời gian thực có thể nhận diện nhiều đối tượng trong cùng một ảnh — nhanh và chính xác. Không giống như các cách tiếp cận cũ dùng sliding window, YOLO thực hiện tất cả trong một lần xử lý, rất phù hợp cho các ứng dụng theo dõi, giám sát hay tự động hóa.
Trong dự án này, mình dùng YOLOv8 cho một Kafka Consumer. Khi các khung hình được gửi đến Kafka Topic, YOLOv8 lắng nghe dòng dữ liệu, xử lý từng khung, và trả về kết quả gồm bounding box, nhãn đối tượng và độ tin cậy.
Điều mình thích nhất ở YOLOv8 là tính nhanh và linh hoạt. Nó chạy mượt trên cả GPU lẫn CPU. Trong bối cảnh bài toán của mình, mình muốn đẩy nó lên các hệ thống biên, như Jeton Nano, thì mô hình này theo mình nó vẫn chạy ổn định. Bên cạnh đó, thư viện Ultralytics thì cực kỳ dễ dùng. Chỉ với vài dòng code là bạn có thể load model có sẵn, chạy dự đoán và hiển thị kết quả.
Nếu bạn mới bắt đầu với object detection, YOLOv8 là lựa chọn rất đáng thử — hiện đại, có tài liệu tốt, và được xây dựng để phục vụ các tác vụ thời gian thực.
Triển khai mã nguồn
Hình dưới minh họa kiến trúc cơ bản của mô hình truyền sự kiện Apache Kafka:
Ở phần trên, một Kafka Producer sẽ gửi các sự kiện (ở đây là các khung hình video) vào Kafka Cluster, bao gồm một hoặc nhiều broker. Mỗi broker sẽ quản lý nhiều partition, nơi dữ liệu được lưu trữ theo thứ tự, không thể thay đổi.
Ở phía dưới, một Kafka Consumer sẽ lắng nghe dữ liệu từ cluster để xử lý sự kiện — ví dụ như chạy YOLOv8 để phát hiện đối tượng. Cách thiết kế này cho thấy Kafka giúp kết nối real-time giữa producer và consumer theo cách tách biệt và dễ mở rộng.
Bạn có thể xem toàn bộ mã nguồn tại đây (đừng quên cho mình một ⭐ để có thêm động lực bay xa nha :)) ):
GitHub - MinLee0210/kafka-learning
Cài đặt Kafka
Mình dùng Docker để cài Kafka, bạn có thể tham khảo thêm trên trang chủ Kafka.
Đây là tệp docker-compose mình sử dụng:
services: zookeeper: image: confluentinc/cp-zookeeper:7.3.2 environment: ZOOKEEPER_CLIENT_PORT: 2181 ZOOKEEPER_TICK_TIME: 2000 kafka: image: confluentinc/cp-kafka:7.3.2 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
Khởi động với:
docker compose -f <tên tệp docker-compose> up -d
Tắt bằng:
docker compose -f <tên tệp docker-compose> down
Thêm -d để chạy nền (background). Nếu muốn xem log khi chạy nền, dùng:
docker compose -f <tên tệp docker-compose> logs -f
Bên cạnh đó, các thư viện mình dùng trong dự án này bao gồm:
- kafka-python => xử lí Kafka.
- ultralytics => xử lí YOLOv8
- pytubefix =>xử lí tải video youtube.
Cài đặt Publisher
Mục đích của script này là đọc một video — từ file local, webcam, luồng RTSP hoặc link YouTube — và gửi từng khung hình như một thông điệp tới Kafka.
from pathlib import Path
import cv2
import numpy as np
from kafka import KafkaProducer from pytubefix import YouTube
from pytubefix.cli import on_progress def download_youtube_video(url: str, dest: str): yt = YouTube(url, on_progress_callback=on_progress) ys = yt.streams.get_highest_resolution() res = ys.download(output_path=dest) return res def np_to_bytes(arr: np.ndarray): return arr.tobytes() def process_video_source(video_name): match video_name: case str() if video_name.startswith("rtsp://"): print("Running rtsp video") return video_name case str() if video_name.startswith("http"): if "youtube" in video_name: print("Running youtube video") video_name = download_youtube_video(video_name, "./") return video_name else: print("Running http video") return video_name case _: print("Running local video") return video_name def main(vido_name: str, topic: str = "VIDEO_STREAM", bootstrap_servers: list[str] = ["localhost:9092"]): producer = KafkaProducer(bootstrap_servers=bootstrap_servers) video_name = process_video_source(vido_name) cap = cv2.VideoCapture(video_name) print(f"Publishing video {video_name} to topic: {topic}") while True: ret, frame = cap.read() if not ret: break # Encode frame as JPEG and convert to bytes ret2, buffer = cv2.imencode('.jpg', frame) if not ret2: continue frame_bytes = np_to_bytes(buffer) producer.send(topic=topic, value=frame_bytes) cap.release() producer.close() if __name__ == "__main__": video_name = "video_name" main(video_name)
Script sẽ kiểm tra nguồn video là gì. Nếu là link YouTube, nó sẽ tải về. Nếu là stream online, sẽ truyền trực tiếp. Nhờ đó bạn có thể linh hoạt dùng bất kỳ nguồn nào.
Cài đặt Listener
Đây là phía consumer trong pipeline video. Khi producer gửi từng khung hình lên Kafka, script này sẽ lắng nghe và xử lý bằng YOLOv8.
import cv2
import numpy as np
from ultralytics import YOLO
from kafka import KafkaConsumer, KafkaProducer VIDEO_STREAM = "VIDEO_STREAM" model = YOLO("yolov8n.pt") def main( topic: str = VIDEO_STREAM, bootstrap_servers: list[str] = ["localhost:9092"]
): consumer = KafkaConsumer( topic, bootstrap_servers=bootstrap_servers ) print("Listening to topic: ", topic) for msg in consumer: frame = np.frombuffer(msg.value, dtype=np.uint8) frame = cv2.imdecode(frame, cv2.IMREAD_COLOR) if frame is not None: # Run YOLO8 tracking on the frame, persisting tracks between frames results = model.track(frame, persist=True) print(results) # Visualize the results on the frame annotated_frame = results[0].plot() # Display the annotated frame cv2.imshow("YOLOv8 Tracking", annotated_frame) # Break the loop if 'q' is pressed if cv2.waitKey(1) & 0xFF == ord("q"): break else: print("Frame is None") if __name__ == "__main__": main()
Nó sẽ:
- Nhận frame từ Kafka dưới dạng byte,
- Dùng OpenCV giải mã lại thành ảnh,
- Dùng YOLOv8 để theo dõi đối tượng (với ID giữa các frame),
- Vẽ kết quả trực tiếp lên khung hình,
- Hiển thị liên tục như một video real-time.
Kết bài
Trong dự án này, mình đã kết hợp Kafka và YOLOv8 để xây dựng một hệ thống theo dõi đối tượng thời gian thực đơn giản nhưng hiệu quả. Kafka đảm nhận phần stream dữ liệu từ producer đến consumer, còn YOLOv8 thực hiện phát hiện và theo dõi đối tượng.
Cách tiếp cận này giúp hệ thống dễ mở rộng, dễ tùy biến và gần gũi với cách hệ thống AI hoạt động trong thực tế. Bạn có thể thay đổi model, thêm bước xử lý hậu kỳ hoặc gửi kết quả lên dashboard — vì mọi thứ đều được tách biệt rõ ràng.
Nếu bạn đang tìm hiểu về AI hoặc hệ thống thời gian thực, đây là một dự án thực hành tuyệt vời — vừa vui, vừa giúp hiểu rõ hơn về cách dữ liệu luân chuyển trong ứng dụng hiện đại.