Elasticsearch Bài học 48: Đồng bộ MySQL/PostgreSQL sang Elasticsearch realtime bằng Debezium & Kafka (CDC)

Trong thế giới kiến trúc phần mềm hiện đại, Elasticsearch gần như không bao giờ đứng một mình làm Database chính (Source of Truth). Nhiệm vụ lưu trữ lõi luôn thuộc về RDBMS như MySQL hay PostgreSQL.

Vấn đề đau đầu nhất của mọi Backend Engineer là: Làm sao để khi một nhân viên cập nhật giá sản phẩm trên MySQL, cái giá đó lập tức xuất hiện trên Elasticsearch để khách hàng search thấy?

1. Nỗi đau của các phương pháp đồng bộ truyền thống

Trước khi có CDC, chúng ta thường dùng 2 phương pháp sau và cả 2 đều tồn tại những nhược điểm chí mạng:

1. Dual-write (Ghi kép tại tầng Application):

   Trong code Laravel/Golang, mỗi khi có lệnh $product->save(), bạn viết thêm một dòng API call để ném sang Elasticsearch.

   • Nỗi đau: Rất dễ sinh lỗi bất đồng bộ. Nếu MySQL báo OK, nhưng API Elasticsearch bị timeout, database của bạn sẽ bị lệch. Thêm nữa, nếu ai đó vào thẳng MySQL chạy lệnh UPDATE bằng tay (Navicat), Elasticsearch sẽ vĩnh viễn không bao giờ biết.

2. Cronjob Polling (Quét định kỳ):

   Bạn viết một script chạy 5 phút/lần: SELECT * FROM products WHERE updated_at > last_run_time. Sau đó bốc đống dữ liệu đó ném sang Elasticsearch bằng Bulk API.

   • Nỗi đau: Tốc độ không phải là Real-time (Độ trễ 5 phút). Quét liên tục làm rác CPU của MySQL. Và quan trọng nhất: Không thể bắt được sự kiện DELETE cứng (Nếu một dòng bị xóa khỏi MySQL, lệnh SELECT sẽ không thấy nó đâu để mà qua Elasticsearch xóa tương ứng).

2. Change Data Capture (CDC) và Debezium là gì?

CDC (Bắt giữ dữ liệu thay đổi) là một mô hình kiến trúc giúp theo dõi trực tiếp mọi thao tác INSERT, UPDATE, DELETE từ tận "đáy" của Database.

Thay vì quét bảng dữ liệu, CDC đọc thẳng vào các file log nội bộ của Database:

• Với MySQL: Nó đọc file Binlog (Binary Log).

• Với PostgreSQL: Nó đọc file WAL (Write-Ahead Log).

Debezium là một bộ công cụ mã nguồn mở (thường chạy trên nền Kafka Connect). Nó cắm thẳng một ống hút vào Binlog của MySQL. Bất cứ khi nào MySQL thực thi một giao dịch, Debezium lập tức chụp lại giao dịch đó và biến nó thành một Event (Sự kiện JSON) ném vào Kafka với tốc độ tính bằng mili-giây.

3. Kiến trúc luồng chảy (Data Pipeline) hoàn hảo

Để thiết lập một luồng CDC hoàn chỉnh, bạn cần 4 thành phần kết nối với nhau như những mắc xích:

MySQL $\rightarrow$ Debezium $\rightarrow$ Kafka $\rightarrow$ Kafka Connect (Elasticsearch Sink) $\rightarrow$ Elasticsearch

• Bước 1: Dữ liệu gốc được ghi vào bảng products trên MySQL.

• Bước 2: MySQL tự động ghi hành động này vào file binlog.

• Bước 3: Debezium Source Connector liên tục đọc binlog. Nó nhìn thấy dòng thay đổi, lập tức đóng gói thành một thông điệp chứa trạng thái before (dữ liệu cũ) và after (dữ liệu mới).

• Bước 4: Debezium ném thông điệp đó vào một Topic trong Kafka (Ví dụ topic tên là: mysql.inventory.products). Tại Kafka, thông điệp này được lưu trữ an toàn, đảm bảo không bao giờ thất thoát dù mạng chập chờn.

• Bước 5: Kafka Connect (Elasticsearch Sink) đứng ở đầu kia, đóng vai trò là Consumer. Nó liên tục "hút" các thông điệp từ Topic, tự động convert chúng thành các lệnh Index, Update, Delete và dùng Bulk API bắn thẳng vào Elasticsearch.

4. Tại sao kiến trúc này lại là Tiêu chuẩn Công nghiệp (Industry Standard)?

Mô hình CDC + Kafka mang lại những lợi ích vô song mà các hệ thống lớn không thể từ chối:

1. Hiệu năng MySQL được giải phóng 100%: Việc đọc Binlog diễn ra dưới tầng OS, cực kỳ nhẹ nhàng. Database của bạn không phải chịu bất kỳ câu query SELECT nào từ các script đồng bộ định kỳ.

2. Bắt trọn mọi thay đổi (Kể cả Delete cứng): Khi một row bị xóa, Binlog sẽ ghi lại hành động xóa. Debezium lập tức ném một cục JSON có trường "op": "d" (Delete) vào Kafka. Sink Connector đọc được sẽ gọi hàm DELETE tương ứng bên Elasticsearch. Dữ liệu sạch sẽ tuyệt đối.

3. Decoupling (Tách rời hệ thống): Giả sử một ngày Elasticsearch bị sập mất 2 tiếng. Với phương pháp Dual-write, API của bạn sẽ bị rớt đơn hàng. Với Kafka đứng ở giữa, dữ liệu vẫn cứ dồn vào Kafka một cách an toàn. Khi Elasticsearch sống lại, Sink Connector sẽ tự động hút tiếp phần tồn đọng. Không thất thoát 1 byte dữ liệu.

5. Lưu ý thực chiến cho Backend Engineer

• Chỉ đồng bộ những gì cần tìm kiếm: Đừng cấu hình Debezium đẩy toàn bộ 100 cột của bảng MySQL sang Elasticsearch. Hãy cấu hình filter ở Kafka Connect để chỉ đẩy qua những cột phục vụ Search/Filter (như name, price, status). Những trường như html_content hay admin_notes hãy giữ lại ở MySQL.

• Xử lý Flatten Data: Dữ liệu từ Binlog là dữ liệu phẳng (Flat). Nếu Elasticsearch của bạn cần một cục JSON Nested (Ví dụ: Order kèm theo mảng Order_Items), Debezium thuần túy không thể làm được. Bạn sẽ phải viết thêm một tầng trung gian (ví dụ Kafka Streams, hoặc một worker Golang) để "gộp" dữ liệu trước khi ném vào Elasticsearch.

CDC là phương pháp tuyệt vời nhất để đồng bộ dữ liệu Business. Tuy nhiên, nếu bạn không muốn đồng bộ Database mà muốn đẩy thẳng Log từ các file text (như Nginx log, hoặc Log từ Docker container của Microservices) vào Elasticsearch thì sao? Mời bạn bước sang Bài 49 với cỗ máy Logstash & Filebeat.