Đã đăng vào khoảng 13 giờ trước 6 phút đọc

Elasticsearch Bài học 13: Bulk API - Tuyệt chiêu Index/Update hàng vạn bản ghi mỗi giây tối ưu

đang viết một worker bằng Golang hoặc PHP để đồng bộ 10 triệu bản ghi sản phẩm từ MySQL sang Elasticsearch. Nếu bạn dùng một vòng lặp for và gọi API POST /products/_doc 10 triệu lần, hệ thống của bạn sẽ mất hàng chục giờ, thậm chí vài ngày mới chạy xong.

Lý do không nằm ở CPU hay RAM, mà nằm ở Network Round-Trip Time (Độ trễ mạng). Mỗi HTTP request gửi đi đều tốn thời gian thiết lập kết nối TCP, gửi header, chờ server xử lý, và nhận response. Nếu độ trễ mạng là 10ms, 10 triệu request tuần tự sẽ tốn 100.000 giây (gần 28 tiếng) chỉ để... chờ mạng.

Giải pháp cho bài toán này là gom (batch) hàng nghìn thao tác vào chung một HTTP Request duy nhất. Đó chính là Bulk API.

1. Bulk API và Cú pháp NDJSON "kỳ lạ"

Khác với các API thông thường nhận payload là một object JSON hoặc mảng JSON [...], Bulk API sử dụng một định dạng đặc biệt gọi là NDJSON (Newline Delimited JSON).

Một request Bulk bắt buộc tuân theo cấu trúc ghép cặp dòng (trừ lệnh delete):

Dòng 1: Metadata (Chứa hành động gì? Index nào? ID bao nhiêu?)
Dòng 2: Dữ liệu thực tế của Document.
Dòng 3: Metadata của thao tác tiếp theo...

Ví dụ thực tế:

JSON

POST _bulk
{"index": {"_index": "products", "_id": "1"}}
{"name": "Sữa rửa mặt A", "price": 100000}
{"update": {"_index": "products", "_id": "2"}}
{"doc": {"price": 150000}}
{"delete": {"_index": "products", "_id": "3"}}

(Lưu ý: Bắt buộc phải có một dấu xuống dòng \n ở cuối cùng của request).

Tại sao không dùng Mảng JSON `[{}, {}]` cho đơn giản?

Đây là một thiết kế kiến trúc thiên tài của Elasticsearch: Nếu bạn gửi một mảng JSON khổng lồ chứa 10.000 phần tử, Coordinating Node sẽ phải nạp toàn bộ chuỗi text đó vào RAM, cấp phát bộ nhớ để parse (phân tích cú pháp) nó thành một cây JSON Object khổng lồ. Việc này vắt kiệt Heap Memory của Java và dễ gây ra lỗi OutOfMemory (OOM).

Với NDJSON, Coordinating Node chỉ cần đọc luồng dữ liệu (Stream) từng dòng một. Đọc đến dòng nào, nó phân loại dòng đó thuộc về Shard nào rồi đẩy đi ngay lập tức mà không cần tốn RAM để parse toàn bộ dữ liệu.

2. Bốn hành động (Operations) trong Bulk API

Bên trong một request Bulk, bạn có thể trộn lẫn hàng ngàn hành động khác nhau. Có 4 loại hành động:

index: Nếu ID đã tồn tại thì ghi đè toàn bộ (giống PUT). Nếu chưa có thì tạo mới. Đây là lệnh được dùng nhiều nhất khi đồng bộ dữ liệu.
create: Chỉ tạo mới. Nếu ID đã tồn tại, nó sẽ quăng lỗi 409 Conflict cho riêng dòng đó. (Rất tốt để chống ghi đè nhầm).
update: Cập nhật một phần Document (Partial Update). Yêu cầu dòng dữ liệu đi kèm phải bọc trong object {"doc": {...}}.
delete: Xóa Document. Lệnh này đặc biệt ở chỗ nó không cần dòng dữ liệu thứ 2 đi kèm, vì chỉ cần ID là đủ xóa.

3. Luồng xử lý phân tán của Bulk API

Khi bạn gửi một file NDJSON chứa 10.000 thao tác lên Coordinating Node, hệ thống xử lý như sau:

Phân tách (Split): Node điều phối đọc metadata từng dòng, dùng thuật toán Hash (đã học ở Bài 9) để tính xem ID này thuộc về Shard nào.
Gom nhóm (Group): Nó tự động gom 10.000 thao tác này thành các gói tin nhỏ hơn tương ứng với các Shard. Ví dụ: 3000 lệnh gửi về Shard 0 (ở Node 1), 7000 lệnh gửi về Shard 1 (ở Node 2).
Thực thi song song (Parallel Execution): Các gói tin này được bắn song song qua mạng nội bộ tới các Data Node.
Thu thập (Gather): Coordinating Node gom kết quả trả về từ các Node, gộp thành một mảng JSON response và trả về cho ứng dụng của bạn.

4. Cạm bẫy "200 OK" và Cách xử lý lỗi

Khi bạn gọi Bulk API, HTTP Status Code trả về thường xuyên là 200 OK. Nhưng điều đó không có nghĩa là tất cả thao tác đều thành công.

Trừ khi cú pháp request bị sai bét, nếu không thì Elasticsearch luôn trả về 200, cho dù bên trong có 9.999 thao tác thành công và 1 thao tác bị lỗi (ví dụ: lỗi Version Conflict).

API Backend của bạn bắt buộc phải kiểm tra trường "errors" ở đầu response:

JSON

{
  "took": 150,
  "errors": true, 
  "items": [
    {
      "index": {
        "_index": "products",
        "_id": "1",
        "status": 201
      }
    },
    {
      "update": {
        "_index": "products",
        "_id": "2",
        "status": 409,
        "error": {
          "type": "version_conflict_engine_exception",
          "reason": "..."
        }
      }
    }
  ]
}

Quy trình chuẩn trong Backend: Parse response, check if (response.errors == true), sau đó lặp qua mảng items để trích xuất ra những ID có status >= 400 và đẩy chúng vào Dead Letter Queue (như Kafka hoặc RabbitMQ) để xử lý lại (Retry).

5. Best Practices: Kích thước Bulk bao nhiêu là chuẩn?

Đừng nhồi 1 triệu bản ghi vào 1 request Bulk. Nó sẽ làm quá tải Network Buffer và RAM của Node điều phối.

Không có con số thần thánh, nhưng đây là tiêu chuẩn được cộng đồng kỹ sư kiểm chứng:

Theo số lượng: Dao động từ 1.000 đến 5.000 Document mỗi request.
Theo dung lượng: Dao động từ 5MB đến 15MB (tối đa không nên vượt quá 50MB) mỗi request.

Cách đo lường thực tế: Nếu bạn dùng Golang/PHP, hãy viết script chạy benchmark. Thử tăng dần kích thước batch (1000 -> 2000 -> 5000). Nếu thấy thời gian phản hồi (Latency) bắt đầu tăng vọt hoặc log Elasticsearch báo warning về Garbage Collector (GC), thì lùi lại một nấc.

Elasticsearch