Mini-challenge — top-K phần tử trên stream

Hệ thống streaming của một nền tảng video cần báo cáo top-10 video được xem nhiều nhất trong giờ vừa qua. Stream đầu vào là hàng chục triệu sự kiện (videoId, timestamp) mỗi giờ — không vừa RAM nếu đếm chính xác toàn bộ. Bài toán xuất hiện ở nhiều dạng thực tế:

• Network monitoring — top-K IP gửi traffic nhiều nhất trong 1 phút (phát hiện DDoS).
• Search engine — top-K query được gõ nhiều nhất trong ngày (autocomplete ranking).
• E-commerce — top-K sản phẩm được click nhiều nhất trong giờ cao điểm (homepage widget).

Dành 20–25 phút tự implement trước khi xem gợi ý.

🎯 Đề bài

Viết ba hàm/cấu trúc:

Cấu trúc CountMinSketch(d, w)

• Input: d hàng (số hàm hash), w cột (độ rộng mỗi hàng).
• Bất biến: với mọi phần tử x, estimate(x) luôn lớn hơn hoặc bằng tần suất thực của x.

Hàm update(cms, x)

• Tăng counter của x trong tất cả d hàng của CMS.

Hàm topK(stream, K, d, w)

• Input: stream các phần tử (đọc tuần tự một lần), số nguyên K, kích thước CMS (d, w).
• Output: danh sách K phần tử có tần suất ước lượng cao nhất, theo thứ tự giảm dần.
• Ràng buộc bộ nhớ: O(d×w + K) — không được lưu toàn bộ stream vào mảng.

Ví dụ: stream = [A, B, A, C, A, B, D, A, B], K = 2 → output [(A, 4), (B, 3)].

🔍 I-P-O

📦 Concept mapping

📋 Test cases

Lưu ý: với CMS, output là xấp xỉ — estimate(x) >= freq(x) luôn đúng nhưng có thể lớn hơn. Test case chính xác khi w đủ lớn (ít collision).

▶️ Starter pseudocode

structure CountMinSketch:
    d <- số hàng (số hàm hash)
    w <- số cột (độ rộng)
    table[d][w] <- 0  -- tất cả counter = 0
    -- TODO: chọn d hàm hash độc lập h_1, ..., h_d

function update(cms, x):
    -- TODO: với mỗi hàng i từ 0 đến d-1:
    --         tăng cms.table[i][h_i(x) mod w]

function estimate(cms, x):
    -- TODO: trả về min của cms.table[i][h_i(x) mod w] với mọi i

function topK(stream, K, d, w):
    cms <- CountMinSketch(d, w)
    heap <- MinHeap kích thước K  -- (tần_suất, phần_tử)
    seen <- Set                    -- các phần tử đã thêm vào heap
    -- TODO: với mỗi phần tử x trong stream:
    --         update(cms, x)
    --         freq <- estimate(cms, x)
    --         cập nhật heap nếu x nên ở trong top-K
    return heap theo thứ tự giảm dần

💡 Gợi ý

✅ Lời giải

structure CountMinSketch(d, w):
    table[d][w] <- 0
    -- d cặp hệ số (a_i, b_i) cho hàm hash pairwise independent
    for i from 0 to d-1:
        a[i] <- random odd number
        b[i] <- random number

function hashRow(cms, i, x):
    -- hash pairwise independent: (a * hash(x) + b) mod p mod w
    -- p là số nguyên tố lớn hơn kích thước universe
    return (cms.a[i] * hash(x) + cms.b[i]) mod PRIME mod cms.w

function update(cms, x):
    for i from 0 to cms.d - 1:
        col <- hashRow(cms, i, x)
        cms.table[i][col] <- cms.table[i][col] + 1
// Time: O(d)  Space: O(d × w)

function estimate(cms, x):
    minVal <- +infinity
    for i from 0 to cms.d - 1:
        col <- hashRow(cms, i, x)
        minVal <- min(minVal, cms.table[i][col])
    return minVal
// Time: O(d)

function topK(stream, K, d, w):
    cms <- CountMinSketch(d, w)
    heap <- MinHeap()            -- heap (freq, element), so sánh theo freq
    inHeap <- Map()              -- element -> freq đang trong heap

    for each element x in stream:
        update(cms, x)
        freq <- estimate(cms, x)

        if x in inHeap:
            -- cập nhật: xóa entry cũ, thêm lại với freq mới
            heap.remove((inHeap[x], x))
            heap.push((freq, x))
            inHeap[x] <- freq
        else if heap.size < K:
            heap.push((freq, x))
            inHeap[x] <- freq
        else if freq > heap.min():
            -- loại phần tử nhỏ nhất, thêm x vào
            (minFreq, minElem) <- heap.pop()
            delete inHeap[minElem]
            heap.push((freq, x))
            inHeap[x] <- freq

    -- trả về K phần tử theo thứ tự giảm dần
    result <- []
    while heap không rỗng:
        result.prepend(heap.pop())
    return result
// Time: O(n × d) + chi phí cập nhật heap  Space: O(d × w + K)

🎓 Mở rộng

So sánh với đếm chính xác — bộ nhớ:

Lossy counting (Manku & Motwani 2002): thuật toán khác cho heavy hitters với bảo đảm lý thuyết khác — thay vì over-estimate, nó có thể bỏ sót phần tử dưới ngưỡng tần suất thấp. Phù hợp khi muốn giới hạn false positive thay vì false negative.

Streaming top-K trong thực tế: Apache Flink dùng CMS + heap cho topN operator trên bounded stream. Twitter Algebird cung cấp TopK monoid có thể merge từ nhiều partition song song — phù hợp với kiến trúc phân tán.

✨ Điều bạn vừa làm được

• Implement Count-Min Sketch d × w với d hàm hash độc lập và bất biến over-estimate.
• Kết hợp CMS với min-heap K phần tử để giải bài top-K heavy hitters trong O(n × d × log K) và O(d×w + K) bộ nhớ — hằng số theo n.
• Hiểu trade-off: chính xác tuyệt đối cần O(n unique) bộ nhớ, xấp xỉ với sai số có giới hạn chỉ cần O(d×w).
• Nhận ra khi nào CMS không phù hợp: khi K lớn so với tổng phần tử, hoặc khi cần biết chính xác tần suất mỗi phần tử.

Bài tiếp theo: Case study — Redis HLL & Kafka offset