Elasticsearch 5 — Indexing & nạp dữ liệu

Ba bài trước tập trung vào phía đọc: mô hình dữ liệu, Query DSL và Aggregations. Nhưng trước khi tìm kiếm được, dữ liệu phải vào cluster. Với một team dữ liệu ngân hàng, đây là phần dễ làm sai nhất về mặt vận hành: nạp từng document một cho hàng triệu bản ghi/ngày sẽ giết cluster, không hiểu vòng đời ghi thì không giải thích được "tại sao index xong chưa tìm thấy" hay "tại sao đĩa đầy vì segment", và chọn sai đường nạp (Beats vs Logstash vs Kafka) khiến pipeline vừa chậm vừa khó bảo trì. Bài này đi từ cơ chế ghi một document, đến _bulk, vòng đời segment, rồi toàn cảnh các đường nạp và cách tối ưu.

Ghi document: index / update / delete

Mọi thao tác ghi đều đi qua HTTP tới cluster (JSON API, không phải SQL). Ba thao tác cơ bản:

Index (tạo hoặc ghi đè theo _id). Đây là thao tác hay dùng nhất khi nạp log/giao dịch:

PUT /transactions-2026.06/_doc/TXN-2026-0009812
{
  "txn_id": "TXN-2026-0009812",
  "account_no": "0501234567",
  "amount": 15000000,
  "currency": "VND",
  "channel": "MOBILE",
  "status": "SUCCESS",
  "created_at": "2026-06-30T09:41:22+07:00"
}

Nếu _id đã tồn tại, ES ghi đè (thực chất là đánh dấu bản cũ đã xoá rồi ghi bản mới, _version tăng). Chính vì thế dùng txn_id làm _id là cách chống trùng lặp cực mạnh khi nạp lại (replay) log — nạp cùng một giao dịch hai lần không sinh hai bản ghi. Nếu muốn ES tự sinh _id ngẫu nhiên thì dùng POST /index/_doc (không kèm id) — nhanh hơn chút vì bỏ qua bước kiểm tra tồn tại, nhưng mất khả năng idempotent.

Update (cập nhật một phần). ES không sửa tại chỗ; nó đọc _source cũ, gộp thay đổi, rồi ghi lại toàn bộ document mới:

POST /transactions-2026.06/_update/TXN-2026-0009812
{ "doc": { "status": "REVERSED" } }

Vì mỗi update = đọc + ghi lại cả document, update nhiều lần liên tục lên cùng một document rất tốn kém; ES dùng _seq_no/_primary_term để tránh xung đột (optimistic concurrency control). Trong tình huống log/giao dịch bất biến, nên nghĩ theo hướng "append" thay vì "update".

Delete. DELETE /transactions-2026.06/_doc/TXN-... chỉ đánh dấu xoá (tombstone). Document vẫn nằm trong segment cho tới khi merge dọn đi (nói ở dưới). Đây là lý do xoá không giải phóng đĩa ngay.

_bulk API — bắt buộc cho khối lượng lớn

Nạp từng document bằng một HTTP request riêng cho mỗi bản ghi là sai lầm kinh điển: overhead mạng + xử lý request lấn át chi phí index thực. Với hàng triệu giao dịch/ngày, phải dùng _bulk: gộp nhiều thao tác vào một request.

Định dạng body của _bulk là NDJSON (mỗi dòng một JSON, phân tách bằng ký tự xuống dòng), theo cặp: một dòng action metadata rồi một dòng document (delete thì chỉ có dòng action):

POST /_bulk
{ "index": { "_index": "transactions-2026.06", "_id": "TXN-1" } }
{ "account_no": "0501234567", "amount": 15000000, "status": "SUCCESS", "created_at": "2026-06-30T09:41:22+07:00" }
{ "index": { "_index": "transactions-2026.06", "_id": "TXN-2" } }
{ "account_no": "0509999888", "amount": 320000, "status": "FAILED", "created_at": "2026-06-30T09:41:23+07:00" }
{ "update": { "_index": "transactions-2026.06", "_id": "TXN-1" } }
{ "doc": { "status": "REVERSED" } }
{ "delete": { "_index": "transactions-2026.06", "_id": "TXN-0" } }

Lưu ý sống còn: mỗi dòng phải kết thúc bằng \n, kể cả dòng cuối, và toàn bộ file gửi với header Content-Type: application/x-ndjson.

Đặc điểm quan trọng của _bulk:

• Từng thao tác độc lập. Một document lỗi (sai mapping) không làm hỏng cả lô; response trả về mảng items[], mỗi phần tử có status riêng. Bạn phải đọc response.errors để biết lô có phần nào fail — đây là lỗi hay bị bỏ sót (client tưởng thành công vì HTTP 200 nhưng bên trong có document bị mapper_parsing_exception).
• Không phải transaction. _bulk không phải giao dịch ACID nhiều-document; không có rollback. Nó chỉ là "gộp request cho hiệu quả".
• Kích thước lô (bulk size) cần tinh chỉnh: quá nhỏ thì overhead vẫn cao, quá lớn (vài chục MB, hàng vạn doc) gây áp lực bộ nhớ và timeout. Điểm khởi đầu thực chiến: 5–15 MB mỗi lô hoặc 1.000–5.000 document, rồi đo và điều chỉnh theo phần cứng.

Vòng đời ghi: buffer → refresh → flush → merge

Đây là phần cốt lõi để hiểu hành vi của ES. Bên dưới là Lucene, với đặc tính segment bất biến (immutable segment). Một document đi qua các giai đoạn sau:

1. Buffer (bộ đệm trong RAM). Khi index một document, nó vào một in-memory buffer. Lúc này document chưa tìm thấy được qua search. Đồng thời, thao tác được ghi ngay vào translog (xem mục sau) để không mất khi crash.

2. Refresh — tạo segment, near-real-time. Định kỳ (mặc định refresh_interval = 1 giây), ES gom buffer thành một segment mới và mở nó cho tìm kiếm. Segment này ban đầu nằm trong OS filesystem cache (chưa fsync xuống đĩa) nhưng đã searchable. Đây chính là lý do ES là near-real-time: document nạp xong cần tối đa ~1s mới truy vấn thấy. Refresh tốn kém (mỗi lần tạo một segment nhỏ); nạp lớn mà giữ refresh 1s sẽ sinh vô số segment nhỏ.

3. Flush — bền vững hoá. Flush thực hiện fsync các segment xuống đĩa và xoá (cắt) translog đã được bền vững. Sau flush, dữ liệu chắc chắn nằm trên đĩa và translog được làm mới. Flush xảy ra tự động (theo kích thước translog hoặc thời gian), bạn hiếm khi gọi thủ công.

4. Merge — hợp nhất segment. Vì segment bất biến, mỗi refresh sinh thêm một segment; update/delete tạo tombstone. Nếu để nguyên, số segment tăng vô hạn làm search chậm và tốn tài nguyên. Lucene chạy merge ở nền: gộp nhiều segment nhỏ thành segment lớn hơn, và trong quá trình đó mới thực sự loại bỏ document đã xoá/ghi đè. Merge dùng CPU và I/O đĩa; đây là lý do xoá document không trả lại đĩa ngay và tại sao nạp nặng lại "nóng" ổ đĩa.

Tóm tắt để nhớ: buffer = chưa thấy; refresh (~1s) = thấy được (near-real-time); flush = bền vững trên đĩa; merge = dọn dẹp + tối ưu đọc.

Translog — vai trò durability

Vấn đề: refresh chỉ đưa segment vào OS cache, còn flush (fsync) thì tốn kém nên không thể làm sau mỗi document. Vậy giữa hai lần flush, nếu node crash thì các document đã index (đang trong buffer/segment chưa fsync) có mất không? Câu trả lời là không — nhờ translog (transaction log).

• Mỗi thao tác ghi được append vào translog trước/song song với việc vào buffer. Mặc định translog được fsync sau mỗi request (index.translog.durability: "request"), nên khi client nhận 200 là thao tác đã bền vững trong translog.
• Khi node khởi động lại sau crash, ES replay translog để dựng lại các document chưa kịp flush thành segment. Không mất dữ liệu đã xác nhận.
• Flush chính là thời điểm "chốt sổ": fsync segment và cắt translog (vì phần đó đã nằm an toàn trong segment trên đĩa).

Có thể đổi durability sang "async" (fsync translog định kỳ, ví dụ mỗi 5s) để nạp nhanh hơn đáng kể, đổi lại chấp nhận rủi ro mất tối đa vài giây dữ liệu nếu crash. Với log giám sát hệ thống có thể chấp nhận; với giao dịch tài chính thì giữ "request". Đây là một quyết định đánh đổi durability–throughput cần lấy ý kiến nghiệp vụ, tương tự cách Kafka đánh đổi acks giữa độ bền và tốc độ.

Các đường nạp dữ liệu vào Elasticsearch

Bạn hiếm khi tự viết code gọi _bulk. Hệ sinh thái Elastic và cộng đồng cung cấp nhiều "đường ống" chuẩn. Chọn đúng đường quyết định độ phức tạp vận hành.

Beats — thu gom nhẹ tại nguồn

Beats là các agent nhẹ (Go, ít RAM) chạy ngay trên máy nguồn, đẩy dữ liệu đi:

• Filebeat — đọc/tail file log (ứng dụng, nginx, syslog), theo dõi vị trí đọc (offset) để không đọc trùng khi restart, và tự động dùng _bulk dưới nền. Có backpressure: nếu ES chậm, Filebeat chậm lại thay vì mất log.
• Metricbeat — thu thập số liệu hệ thống/dịch vụ (CPU, RAM, đĩa, JVM, DB) theo chu kỳ.
• Các beat khác: Packetbeat (traffic mạng), Auditbeat (sự kiện bảo mật/audit), Heartbeat (uptime).

Beats phù hợp khi cần thu tại chỗ, nhẹ, có sẵn module dựng dashboard nhanh. Beats có thể đẩy thẳng vào ES (kèm ingest pipeline) hoặc qua Logstash/Kafka nếu cần xử lý nặng.

Logstash — parse / transform / enrich

Logstash là công cụ ETL luồng, kiến trúc input → filter → output, mạnh về biến đổi phức tạp mà ingest pipeline khó làm: parse đa định dạng, tra cứu làm giàu dữ liệu (lookup DB/CSV), routing có điều kiện, buffer trên đĩa (persistent queue). Đổi lại nó nặng hơn Beats (chạy trên JVM). Mẫu cấu hình rút gọn:

# Logstash pipeline (minh hoạ, KHÔNG phải SQL)
input  { beats { port => 5044 } }
filter {
  grok  { match => { "message" => "%{COMBINEDAPACHELOG}" } }
  date  { match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ] }
  geoip { source => "clientip" }
}
output { elasticsearch { hosts => ["es:9200"] index => "web-%{+YYYY.MM.dd}" } }

Mẫu triển khai phổ biến: Filebeat (thu) → Logstash (xử lý) → Elasticsearch. Beats nhẹ ở nguồn, còn xử lý nặng tập trung ở Logstash.

Ingest pipeline — xử lý ngay trong ES

Nếu chỉ cần biến đổi vừa phải, không cần Logstash: dùng ingest pipeline — chuỗi processor chạy trên ingest node của chính cluster trước khi document được index. Định nghĩa pipeline một lần, rồi trỏ index/bulk vào nó:

PUT /_ingest/pipeline/txn-enrich
{
  "processors": [
    { "grok": {
        "field": "message",
        "patterns": ["%{TIMESTAMP_ISO8601:ts} %{WORD:status} amount=%{NUMBER:amount:float}"]
    }},
    { "date":  { "field": "ts", "target_field": "created_at",
                 "formats": ["ISO8601"], "timezone": "Asia/Ho_Chi_Minh" }},
    { "geoip": { "field": "client_ip", "target_field": "geo" }},
    { "remove": { "field": "message" }}
  ]
}

Các processor thường dùng: grok (tách trường từ chuỗi log theo pattern), date (parse chuỗi thời gian thành field date, ép múi giờ), geoip (tra IP → quốc gia/thành phố/toạ độ), ngoài ra set, rename, remove, convert, script... Nạp dữ liệu qua pipeline:

POST /transactions-2026.06/_doc?pipeline=txn-enrich
{ "message": "2026-06-30T09:41:22 SUCCESS amount=15000000", "client_ip": "1.53.0.10" }

Ưu điểm: không thêm hạ tầng ngoài cluster, đủ cho phần lớn nhu cầu "log có cấu trúc đơn giản". Nhược điểm: xử lý nặng sẽ ăn CPU của chính cluster search; grok phức tạp/nhiều lookup vẫn nên đẩy sang Logstash.

Fluentd / Fluent Bit và các nguồn khác

Ngoài hệ Elastic còn có Fluentd và Fluent Bit — bộ thu/định tuyến log phổ biến, đặc biệt trong Kubernetes (thường chạy dạng DaemonSet gom log container rồi ghi vào ES). Fluent Bit rất nhẹ (viết bằng C), Fluentd giàu plugin hơn. Về vai trò, chúng tương tự Beats/Logstash: thu, buffer, biến đổi nhẹ, rồi output ra ES qua _bulk.

Nạp từ Kafka

Trong kiến trúc dữ liệu ngân hàng, Kafka thường là "xương sống" trung chuyển sự kiện: CDC từ core banking hay log giao dịch đổ vào topic Kafka, và ES chỉ là một trong nhiều consumer. Hai cách phổ biến đưa Kafka → ES:

• Elasticsearch Sink Connector (chạy trên Kafka Connect): đọc topic, ghi vào ES theo lô. Cấu hình khai báo, có sẵn cơ chế mapping key→_id, retry, dead-letter.
• Logstash kafka input: Logstash làm consumer, xử lý rồi output ra ES.

Lợi ích lớn của việc chèn Kafka vào giữa: decouple nguồn và ES (ES bảo trì/downtime thì dữ liệu vẫn nằm an toàn trong Kafka, replay lại sau), hấp thụ đỉnh tải (buffer), và cho phép nhiều đích dùng chung một luồng (ES + kho lakehouse + hệ giám sát gian lận).

Thiết kế index cho dữ liệu chuỗi thời gian

Log và giao dịch là time-series (chỉ append, gắn timestamp). Đừng nhét tất cả vào một index khổng lồ; hãy chia index theo thời gian:

• Index theo ngày/tháng với tên như transactions-2026.06.30 hoặc logs-app-2026.06. Lợi ích: xoá dữ liệu cũ = drop cả index (rất rẻ, khác hẳn DELETE by query phải sinh tombstone + merge); truy vấn theo khoảng thời gian chỉ chạm vào các index liên quan; dễ áp chính sách lưu trữ (retention).
• Data stream — trừu tượng hoá hiện đại của Elastic cho time-series: bạn ghi vào một tên logic (ví dụ logs-txn), ES tự tạo các backing index ẩn (.ds-logs-txn-...) và tự cuộn (rollover) sang index mới khi đủ lớn/đủ lâu. Data stream chỉ nhận append (index doc mới), phù hợp bản chất log/giao dịch bất biến.
• ILM (Index Lifecycle Management) tự động hoá vòng đời: hot (đang nạp, SSD nhanh) → warm (đọc ít, giảm replica) → cold/frozen (lưu trữ rẻ) → delete (xoá theo retention, ví dụ 90 ngày nóng, 1 năm lưu trữ, rồi xoá). Kết hợp với alias/data stream để ứng dụng luôn truy vấn qua một tên bất biến.

Tối ưu tốc độ nạp khối lượng lớn

Khi nạp lô lớn (backfill lịch sử, reindex hàng trăm triệu document), một vài chỉnh sửa tạo khác biệt lớn. Nhớ trả về giá trị cũ sau khi nạp xong.

Ví dụ chỉnh nhanh trước khi backfill và khôi phục sau đó:

PUT /transactions-2026.06/_settings
{ "index": { "refresh_interval": "-1", "number_of_replicas": 0 } }

// ... chạy nạp _bulk ...

PUT /transactions-2026.06/_settings
{ "index": { "refresh_interval": "1s", "number_of_replicas": 1 } }
POST /transactions-2026.06/_refresh

Chi tiết theo dõi và vận hành cluster khi nạp nặng (áp lực merge, disk watermark, thread pool) sẽ nói ở Performance & Ops; phần shard/replica và mở rộng ở Cluster & Scaling.

Use case thực tế

Bối cảnh. Team dữ liệu NCB nạp log giao dịch vào ES để bộ phận Vận hành tra cứu và Giám sát gian lận thống kê. Nguồn là log ứng dụng core (dạng dòng text) trên nhiều máy chủ, khoảng 1,8 triệu giao dịch/ngày (~21 doc/giây trung bình, đỉnh giờ cao điểm gấp 4–5 lần).

Kiến trúc chọn. Filebeat trên từng máy chủ (nhẹ, có offset + backpressure) → Kafka topic txn-log (decouple, hấp thụ đỉnh, cho lakehouse dùng chung) → Logstash consume, dùng grok tách trường + date ép múi giờ Asia/Ho_Chi_Minh + lookup mã chi nhánh → ghi vào data stream logs-txn bằng _bulk. Index chia theo ngày qua data stream + ILM giữ nóng 90 ngày.

Sự cố & xử lý (thực tế hay gặp).

• Ban đầu client tự viết nạp từng document; cluster nghẽn, latency search tăng vọt. Chuyển sang _bulk (lô ~4.000 doc) qua Logstash → throughput tăng nhiều lần.
• Một đợt backfill 3 tháng lịch sử (~162 triệu doc) chạy chậm và làm nóng ổ đĩa vì merge liên tục. Team tạm đặt refresh_interval: "-1" và number_of_replicas: 0, chạy 6 luồng bulk song song, canh không vượt 429; nạp xong mới bật lại refresh 1s và replica 1, rồi _refresh. Thời gian backfill giảm đáng kể.
• Có ngày ES bảo trì 40 phút. Nhờ Kafka giữ dữ liệu, Logstash tự bắt lại từ offset khi ES trở lại — không mất log nào.

Kết quả. Log giao dịch searchable trong ~1 giây (near-real-time nhờ refresh), durability đảm bảo bằng translog request cho luồng giao dịch tài chính; fraud thống kê date_histogram theo giờ với time_zone: "+07:00" cho số đúng; xoá dữ liệu quá 90 ngày = drop backing index (rẻ, không sinh tombstone).

Ghi nhớ

• Ghi = index / update / delete qua JSON API. Index trùng _id là ghi đè → dùng txn_id làm _id để idempotent (chống trùng khi replay). Update = đọc + ghi lại cả document; delete chỉ đánh dấu tombstone.
• _bulk là bắt buộc cho khối lượng lớn: gộp nhiều thao tác (NDJSON, mỗi dòng kết thúc \n). Từng thao tác độc lập, không phải transaction — phải kiểm tra response.errors. Bulk size khởi điểm 5–15 MB / 1.000–5.000 doc.
• Vòng đời ghi: buffer (chưa thấy) → refresh ~1s (tạo segment, near-real-time, thấy được) → flush (fsync segment, cắt translog, bền vững) → merge (gộp segment, mới thực sự dọn doc đã xoá).
• Translog đảm bảo durability giữa hai lần flush: ghi/fsync ngay theo request (mặc định), replay khi crash. async giúp nạp nhanh nhưng chấp nhận mất vài giây — hợp với log, không hợp giao dịch tài chính.
• Đường nạp: Beats (Filebeat/Metricbeat — nhẹ, tại nguồn), Logstash (parse/transform/enrich nặng), ingest pipeline (grok/date/geoip ngay trong ES, không cần hạ tầng ngoài), Fluentd/Fluent Bit (phổ biến trong K8s). Chèn Kafka ở giữa để decouple, hấp thụ đỉnh, nhiều đích dùng chung.
• Time-series: chia index theo ngày/tháng hoặc dùng data stream + ILM — xoá cũ = drop index (rẻ hơn delete-by-query nhiều).
• Tối ưu nạp lớn: tăng refresh_interval (hoặc -1), tạm number_of_replicas: 0, bulk song song vừa phải (canh 429), rồi khôi phục thiết lập sau khi xong.