Full-text, lọc, sắp xếp, phân trang, ElasticsearchFull-text, filters, sorting, pagination, Elasticsearch
Catalog Service ghi sản phẩm vào PostgreSQL. Debezium đọc WAL (hoặc bảng outbox) và đẩy product.changed lên Kafka; Indexer ghi vào OpenSearch theo lô, kèm version. Search API chỉ đọc qua alias products. Chọn kịch bản “Cập nhật sản phẩm” để xem độ trễ đồng bộ, “Reindex sau alias” để xem đổi mapping không gián đoạn.The Catalog Service writes products to PostgreSQL. Debezium reads the WAL (or an outbox table) and publishes product.changed to Kafka; the Indexer bulk-writes to OpenSearch with versions. The Search API only reads through the products alias. Pick “Product updated” to see sync lag, or “Reindex behind an alias” to see a mapping change with no downtime.
1Người dùng gõ “ao khoac nam” (không dấu), chọn thương hiệu Uniqlo và khoảng giá 200k–800k.A user types “ao khoac nam” without diacritics, picks the Uniqlo brand and a 200k–800k price range.
Vài trăm nghìn tới vài triệu dòng, ít facet, chấp nhận ts_rank đơn giản: tsvector + GIN + unaccent là đủ, khỏi đồng bộ hệ thứ hai. Cần facet phức tạp, sửa lỗi gõ, gợi ý, chấm điểm tinh chỉnh hoặc tải tìm kiếm lớn thì chuyển sang OpenSearch. Đặt sau một SearchPort để đổi engine không đụng nghiệp vụ.Hundreds of thousands to a few million rows, few facets, simple ts_rank: tsvector, GIN and unaccent are enough, with no second system to sync. Rich facets, typo tolerance, suggestions, tuned scoring or heavy search load call for OpenSearch. Put it behind a SearchPort so the engine can change without touching business code.
Ghi DB rồi gọi OpenSearch trong request thì một trong hai có thể lỗi. Ghi thay đổi qua outbox hoặc CDC (Debezium đọc WAL), Indexer tiêu thụ từ Kafka và thử lại tới khi thành công.Writing the DB and then calling OpenSearch in the request lets one of them fail alone. Capture changes through an outbox or CDC (Debezium on the WAL); the Indexer consumes from Kafka and retries until it succeeds.
Retry hoặc backfill có thể đưa bản cũ tới sau bản mới. Key Kafka theo id sản phẩm giữ thứ tự trong partition; thêm version_type=external với version của dòng để OpenSearch tự từ chối bản cũ.Retries or a backfill can deliver an old copy after a new one. Keying Kafka by product id keeps order per partition; version_type=external with the row version makes OpenSearch reject the stale copy itself.
Analyzer là chuỗi char filter → tokenizer → token filter. Dùng icu_tokenizer + lowercase + icu_folding (hoặc asciifolding với preserve_original) để “đồng hồ” khớp “dong ho”; giữ thêm field gốc có dấu và boost cao hơn để kết quả đúng dấu lên trước. Lưu ý chữ đ phải gập thành d.An analyzer is a chain: char filters → tokenizer → token filters. Use icu_tokenizer, lowercase and icu_folding (or asciifolding with preserve_original) so “đồng hồ” matches “dong ho”; keep an accented field with a higher boost so exact-diacritic matches rank first. Make sure đ folds to d.
Mỗi bộ lọc là một Specification nhỏ tự biết thêm term, range hay terms vào filter. Builder ghép chúng với must, should, function_score và aggregation; facet của chính bộ lọc đang chọn đặt vào post_filter để vẫn đếm đủ các lựa chọn khác.Each filter is a small Specification that knows how to add its term, range or terms clause. A Builder combines them with must, should, function_score and aggregations; the selected facet's own filter goes into post_filter so its other options still get counts.
from + size buộc mỗi shard xếp hạng from+size tài liệu; OpenSearch chặn ở max_result_window 10.000. Dùng search_after với sort có tie-breaker (score, id), kèm point-in-time nếu cần ảnh chụp nhất quán khi cuộn.from + size makes every shard rank from+size documents, and OpenSearch stops at max_result_window 10,000. Use search_after with a tie-breaking sort (score, id), plus a point-in-time when the scroll needs a consistent snapshot.
Client luôn đọc qua alias, không gọi tên index thật. Tạo index mới, ghi kép, backfill, kiểm tra, rồi đổi alias bằng một lệnh _aliases nguyên tử. Alias đóng vai proxy đứng trước index thật.Clients always read through an alias, never a physical index name. Create the new index, dual-write, backfill, verify, then switch with one atomic _aliases call. The alias acts as a proxy in front of the real index.
| BảngTable | Cột chínhKey columns | Ghi chúNotes |
|---|---|---|
product | id, name, brand, category_id, price, in_stock, sold_30d, description, version, updated_at, search_vector tsvector | Nguồn sự thật. version tăng mỗi lần sửa; search_vector là cột generated với GIN index khi chỉ dùng PostgreSQLSource of truth. version increases on every edit; search_vector is a generated column with a GIN index when PostgreSQL alone is enough |
outbox | id, aggregate_id, type, payload, created_at | Dùng khi không bật CDC trên bảng product; Debezium outbox router đẩy sang product.changedUsed when CDC is not enabled on product itself; the Debezium outbox router publishes it to product.changed |
products_v4 (OpenSearch) | id keyword, name text(vi_folding)+name.raw, brand text+brand.keyword, category keyword, price scaled_float, in_stock boolean, sold_30d integer | Mapping strict (dynamic: strict) để field lạ bị từ chối thay vì tự đoán kiểuStrict mapping (dynamic: strict) so unknown fields are rejected instead of guessed |
alias products / products_write | products → products_v4, products_write → products_v4 (is_write_index) | Search API đọc products, Indexer ghi products_write; reindex chỉ đổi aliasThe Search API reads products, the Indexer writes products_write; a reindex only moves aliases |
Java: Search API ghép filter, chấm điểm, facet và search_after bằng Elasticsearch Java client (OpenSearch client có API gần như giống hệt), kèm phương án PostgreSQL thuần. Go: Indexer ghi bulk có external version.Java: the Search API composes filters, scoring, facets and search_after with the Elasticsearch Java client (the OpenSearch client is nearly identical), plus the plain PostgreSQL alternative. Go: the Indexer bulk-writes with external versions.
@Service
@RequiredArgsConstructor
public class ProductSearchService {
private final ElasticsearchClient es;
public SearchPage<ProductDoc> search(ProductQuery r) throws IOException {
List<Query> filters = new ArrayList<>();
if (r.brand() != null)
filters.add(TermQuery.of(t -> t.field("brand.keyword").value(r.brand()))._toQuery());
if (r.minPrice() != null || r.maxPrice() != null)
filters.add(RangeQuery.of(q -> q.number(n -> n.field("price")
.gte(r.minPrice()).lte(r.maxPrice())))._toQuery());
filters.add(TermQuery.of(t -> t.field("in_stock").value(true))._toQuery());
SearchResponse<ProductDoc> res = es.search(s -> s
.index("products") // alias, not products_v4
.size(24)
.query(q -> q.functionScore(fs -> fs
.query(b -> b.bool(bool -> bool
.must(m -> m.multiMatch(mm -> mm.query(r.text())
.fields("name^3", "name.raw^4", "brand^2", "description")))
.filter(filters)))
.functions(f -> f.fieldValueFactor(v -> v
.field("sold_30d").modifier(FieldValueFactorModifier.Log1p)))
.boostMode(FunctionBoostMode.Sum)))
.sort(o -> o.score(sc -> sc.order(SortOrder.Desc)))
.sort(o -> o.field(f -> f.field("id").order(SortOrder.Asc)))
.searchAfter(r.cursor()) // [score, id] of the last hit
.aggregations("brand", a -> a.terms(t -> t.field("brand.keyword").size(20)))
.aggregations("price", a -> a.range(g -> g.field("price").ranges(
x -> x.to(300_000.0), x -> x.from(300_000.0).to(800_000.0),
x -> x.from(800_000.0)))),
ProductDoc.class);
return SearchPage.of(res);
}
}type ProductChanged struct {
ID string `json:"id"`
Version int64 `json:"version"` // product.version from PostgreSQL
Deleted bool `json:"deleted"`
Doc json.RawMessage `json:"doc"`
}
type Indexer struct {
bulk esutil.BulkIndexer // FlushInterval 1s, FlushBytes 5 MB
targets func() []string // ["products_write"], or v3 + v4 during a reindex
dlq Publisher
}
func (ix *Indexer) Handle(ctx context.Context, m kafka.Message) error {
var ev ProductChanged
if err := json.Unmarshal(m.Value, &ev); err != nil {
return ix.dlq.Publish(ctx, m) // poison message: do not block the partition
}
for _, index := range ix.targets() {
item := esutil.BulkIndexerItem{
Index: index,
Action: "index",
DocumentID: ev.ID,
Version: &ev.Version,
VersionType: "external", // older than the stored version → 409
Body: bytes.NewReader(ev.Doc),
OnFailure: func(ctx context.Context, _ esutil.BulkIndexerItem,
res esutil.BulkIndexerResponseItem, err error) {
if res.Status == http.StatusConflict {
return // a newer version is already indexed
}
_ = ix.dlq.Publish(ctx, m)
},
}
if ev.Deleted {
item.Action, item.Body = "delete", nil
}
if err := ix.bulk.Add(ctx, item); err != nil {
return err
}
}
return nil
}| PostgreSQL tsvector + GINPostgreSQL tsvector + GIN | Elasticsearch / OpenSearchElasticsearch / OpenSearch | |
|---|---|---|
| Hợp vớiGood for | Tới vài triệu dòng, tìm trong admin, ít facetUp to a few million rows, admin search, few facets | Tìm kiếm cho khách hàng, nhiều facet, tải lớnCustomer-facing search, many facets, heavy load |
| Tiếng Việt, lỗi gõVietnamese, typos | unaccent + simple; pg_trgm cho gõ saiunaccent + simple; pg_trgm for typos | ICU folding, fuzziness, synonym, completion suggesterICU folding, fuzziness, synonyms, completion suggester |
| Nhất quánConsistency | Cùng transaction, thấy ngaySame transaction, visible at once | Trễ vài giây, cần pipeline đồng bộSeconds behind, needs a sync pipeline |
| Chi phí vận hànhOperating cost | Không thêm hệ thống nàoNo extra system | Cụm riêng, CDC, Kafka, quy trình reindexA separate cluster, CDC, Kafka, a reindex procedure |
Mỗi pattern mở bài học tương ứng trong menu Pattern.Each one opens its lesson in the Patterns menu.
Khi dữ liệu vài triệu dòng, tải tìm kiếm vừa phải, ít facet và cần thấy ngay sau khi ghi. tsvector generated + GIN + unaccent, thêm pg_trgm cho lỗi gõ. Mỗi hệ thống thêm vào là một pipeline đồng bộ, một cụm phải vận hành và một nguồn lệch dữ liệu.When the data is a few million rows, search load is moderate, facets are few and results must be visible right after a write. A generated tsvector with GIN and unaccent, plus pg_trgm for typos. Every extra system brings a sync pipeline, a cluster to run and a place for data to drift.
Outbox ghi cùng transaction hoặc CDC đọc WAL bằng Debezium, đẩy lên Kafka với key là id. Indexer tiêu thụ, ghi bulk có external version và commit offset sau khi thành công. Định kỳ chạy job so khớp (đếm, checksum theo khoảng id) để bắt lệch còn sót.An outbox written in the same transaction, or CDC on the WAL with Debezium, into Kafka keyed by id. The Indexer consumes, bulk-writes with external versions and commits offsets after success. A periodic reconciliation job (counts, checksums by id range) catches any remaining drift.
Mỗi shard phải xếp hạng from+size tài liệu rồi node điều phối gộp lại, chi phí tăng tuyến tính theo độ sâu; mặc định bị chặn ở 10.000. search_after dùng sort values của kết quả cuối nên mỗi trang tốn như nhau; cần point-in-time nếu muốn ảnh chụp ổn định.Every shard ranks from+size documents and the coordinator merges them, so cost grows with depth, and it is capped at 10,000 by default. search_after uses the last hit's sort values so every page costs the same; add a point-in-time for a stable snapshot.
Không đổi tại chỗ được. Tạo index mới với mapping mới, ghi kép, backfill từ DB hoặc _reindex, kiểm tra số lượng và truy vấn mẫu, đổi alias bằng một lệnh _aliases, giữ index cũ một thời gian để quay lại.It cannot change in place. Create a new index with the new mapping, dual-write, backfill from the DB or with _reindex, check counts and sample queries, switch the alias in one _aliases call, and keep the old index for a while as a rollback.
Phím 1–4 để chọn đáp án, Enter để kiểm tra.Press 1–4 to choose an answer, Enter to check.
Ôn tập ngắt quãng (spaced repetition): thẻ được lặp lại sau 1 → 3 → 7 → 14 → 30 ngày để củng cố trí nhớ dài hạn.Spaced repetition: cards are repeated after 1 → 3 → 7 → 14 → 30 days to build long-term memory.