Module collections — Cấu trúc dữ liệu chuyên biệt

Mỗi khi bạn viết if key not in dict rồi khởi tạo giá trị mặc định, hoặc dùng list.insert(0, x) rồi thắc mắc tại sao hệ thống chậm dần theo thời gian — đó là lúc collections giải quyết vấn đề. Module này cung cấp các container chuyên biệt, được tối ưu ở tầng C, thay thế cho những pattern lặp đi lặp lại với dict và list thông thường.

Trong production, sự khác biệt giữa list.pop(0) — O(n) — và deque.popleft() — O(1) — không phải lý thuyết. Với một message queue xử lý 50.000 events/giây, đó là khoảng cách giữa hệ thống chạy mượt và hệ thống nghẽn cổ chai. Module collections tồn tại vì những kỹ sư CPython đã gặp đúng những bài toán này trong thực tế.

Bài viết này đi sâu vào 6 container chính: Counter, defaultdict, OrderedDict, deque, namedtuple, và ChainMap — kèm theo pipeline xử lý dữ liệu thực tế.

Bức tranh tư duy

Hãy nghĩ collections như bộ dụng cụ chuyên dụng của thợ cơ khí. dict thông thường là chiếc cờ-lê vạn năng — dùng được mọi nơi nhưng không tối ưu chỗ nào. defaultdict là cờ-lê tự chỉnh — không cần kiểm tra kích thước trước khi siết. Counter là cân điện tử — đặt lên là biết trọng lượng ngay. deque là băng chuyền hai chiều — hàng vào ra ở cả hai đầu đều nhanh như nhau.

dict thường:     [kiểm tra key] → [khởi tạo] → [gán giá trị]  (3 bước)
defaultdict:     [gán giá trị]                                  (1 bước, tự khởi tạo)

list:            ← chèn đầu O(n) | chèn cuối O(1) →
deque:           ← chèn đầu O(1) | chèn cuối O(1) →

Counter:         iterable → bảng tần suất (1 dòng code)
ChainMap:        dict_1 + dict_2 + dict_3 → tra cứu theo thứ tự ưu tiên

Điểm mấu chốt: mỗi container giải quyết một pattern cụ thể. Khi bạn nhận ra pattern, chọn đúng container sẽ giảm code và tăng hiệu năng cùng lúc. Nhưng cần cẩn trọng — defaultdict tạo key khi bạn chỉ đọc, và deque trả giá O(n) cho random access. Không có container nào là "tốt nhất", chỉ có container phù hợp nhất.

Cốt lõi kỹ thuật

Counter — Bảng tần suất

Counter là subclass của dict, chuyên đếm tần suất phần tử trong iterable. Kết quả trả về dạng {phần_tử: số_lần_xuất_hiện}.

python

from collections import Counter

# Đếm từ iterable
words = ["apple", "banana", "apple", "cherry", "banana", "apple"]
counter = Counter(words)
# Counter({'apple': 3, 'banana': 2, 'cherry': 1})

# Đếm ký tự
char_count = Counter("mississippi")
# Counter({'i': 4, 's': 4, 'p': 2, 'm': 1})

# Các phương thức quan trọng
counter.most_common(2)        # [('apple', 3), ('banana', 2)]
counter.total()               # 6 (Python 3.10+)
list(counter.elements())      # ['apple', 'apple', 'apple', 'banana', ...]

# Phép toán tập hợp trên Counter
c1 = Counter(a=3, b=1)
c2 = Counter(a=1, b=2)
c1 + c2   # Counter({'a': 4, 'b': 3})  — cộng
c1 - c2   # Counter({'a': 2})           — trừ, bỏ giá trị ≤ 0
c1 & c2   # Counter({'a': 1, 'b': 1})   — min
c1 | c2   # Counter({'a': 3, 'b': 2})   — max

defaultdict — Dict tự khởi tạo

defaultdict nhận một factory function. Khi truy cập key chưa tồn tại, nó tự gọi factory để tạo giá trị mặc định.

python

from collections import defaultdict

# Grouping pattern — phổ biến nhất
users_by_role = defaultdict(list)
for user in users:
    users_by_role[user.role].append(user)

# Counting pattern
error_counts = defaultdict(int)
for log_entry in logs:
    if log_entry.level == "ERROR":
        error_counts[log_entry.service] += 1

# Nested defaultdict — tạo cấu trúc lồng tự động
def nested_dict():
    return defaultdict(nested_dict)

data = nested_dict()
data["users"]["alice"]["settings"]["theme"] = "dark"

OrderedDict — Dict giữ thứ tự chèn

Từ Python 3.7, dict thông thường đã giữ thứ tự chèn. Nhưng OrderedDict vẫn cần thiết cho hai tính năng đặc biệt: move_to_end() và so sánh equality theo thứ tự.

python

from collections import OrderedDict

od = OrderedDict()
od["a"] = 1
od["b"] = 2
od["c"] = 3

od.move_to_end("a")          # Di chuyển "a" xuống cuối
od.move_to_end("c", last=False)  # Di chuyển "c" lên đầu

# OrderedDict so sánh cả thứ tự
OrderedDict(a=1, b=2) == OrderedDict(b=2, a=1)  # False
dict(a=1, b=2) == dict(b=2, a=1)                  # True

# Pattern: LRU Cache đơn giản
class LRUCache(OrderedDict):
    def __init__(self, capacity: int):
        super().__init__()
        self.capacity = capacity

    def get(self, key):
        if key in self:
            self.move_to_end(key)
            return self[key]
        return -1

    def put(self, key, value):
        if key in self:
            self.move_to_end(key)
        self[key] = value
        if len(self) > self.capacity:
            self.popitem(last=False)

deque — Hàng đợi hai đầu

deque (double-ended queue) hỗ trợ thêm/xóa phần tử ở cả hai đầu với O(1). So với list, thao tác ở đầu danh sách nhanh hơn gấp nhiều lần.

python

from collections import deque

d = deque([1, 2, 3])

# Thao tác O(1) ở cả hai đầu
d.append(4)          # Thêm cuối
d.appendleft(0)      # Thêm đầu
d.pop()              # Xóa cuối
d.popleft()          # Xóa đầu

# deque với kích thước cố định — tự loại bỏ phần tử cũ nhất
recent_logs = deque(maxlen=100)
for log in log_stream:
    recent_logs.append(log)  # Luôn giữ 100 logs gần nhất

# Xoay vòng
d = deque([1, 2, 3, 4, 5])
d.rotate(2)   # deque([4, 5, 1, 2, 3]) — xoay phải
d.rotate(-2)  # deque([1, 2, 3, 4, 5]) — xoay trái

Thao tác	`list`	`deque`
`append(x)`	O(1)	O(1)
`pop()`	O(1)	O(1)
`insert(0, x)` / `appendleft(x)`	O(n)	O(1)
`pop(0)` / `popleft()`	O(n)	O(1)
`[i]` random access	O(1)	O(n)

namedtuple — Tuple có tên trường

namedtuple tạo tuple subclass với trường có tên, vừa immutable vừa lightweight.

python

from typing import NamedTuple

# Cách khuyến nghị: dùng typing.NamedTuple
class Coordinate(NamedTuple):
    lat: float
    lon: float
    label: str = ""

    def distance_to(self, other: "Coordinate") -> float:
        return ((self.lat - other.lat)**2 + (self.lon - other.lon)**2) ** 0.5

hcm = Coordinate(10.762622, 106.660172, "TP.HCM")
hn = Coordinate(21.028511, 105.804817, "Hà Nội")

# Truy cập bằng tên hoặc index
print(hcm.lat)      # 10.762622
print(hcm[0])       # 10.762622

# Hashable — dùng làm dict key
locations = {hcm: "Văn phòng HCM", hn: "Văn phòng HN"}

# Unpacking
lat, lon, label = hcm

ChainMap — Tra cứu theo chuỗi ưu tiên

ChainMap nhóm nhiều dict lại, tra cứu theo thứ tự ưu tiên mà không copy dữ liệu.

python

from collections import ChainMap

defaults = {"theme": "dark", "language": "vi", "page_size": 20}
user_prefs = {"theme": "light"}
cli_args = {"page_size": 50}

# Ưu tiên: cli_args > user_prefs > defaults
config = ChainMap(cli_args, user_prefs, defaults)
config["theme"]       # "light"   — từ user_prefs
config["page_size"]   # 50        — từ cli_args
config["language"]    # "vi"      — từ defaults

# Thêm scope mới (ví dụ: override tạm thời)
with_override = config.new_child({"theme": "solarized"})
with_override["theme"]  # "solarized"
config["theme"]         # "light" — không bị ảnh hưởng

Thực chiến

Tình huống: Pipeline xử lý log truy cập web server

Bối cảnh: Hệ thống web server ghi log truy cập, mỗi dòng chứa IP, endpoint, status code. Cần phân tích: top IP truy cập, tỷ lệ lỗi theo endpoint, và giữ 1000 request gần nhất để debug.

Mục tiêu: Xây dựng pipeline phân tích real-time, xử lý lazy, tiết kiệm bộ nhớ.

python

from collections import Counter, defaultdict, deque
from typing import Iterator, NamedTuple
from datetime import datetime


class AccessLog(NamedTuple):
    timestamp: datetime
    ip: str
    endpoint: str
    status_code: int
    response_time_ms: float


class AccessAnalyzer:
    """Phân tích log truy cập web server theo thời gian thực."""

    def __init__(self, recent_buffer_size: int = 1000):
        self.ip_counter = Counter()
        self.errors_by_endpoint = defaultdict(list)
        self.recent_requests = deque(maxlen=recent_buffer_size)
        self._total_requests = 0

    def ingest(self, logs: Iterator[AccessLog]) -> None:
        for log in logs:
            self._total_requests += 1
            self.ip_counter[log.ip] += 1
            self.recent_requests.append(log)

            if log.status_code >= 400:
                self.errors_by_endpoint[log.endpoint].append(
                    (log.timestamp, log.status_code, log.ip)
                )

    def top_ips(self, n: int = 10) -> list[tuple[str, int]]:
        return self.ip_counter.most_common(n)

    def error_rate_by_endpoint(self) -> dict[str, float]:
        endpoint_total = Counter(
            req.endpoint for req in self.recent_requests
        )
        return {
            endpoint: len(errors) / max(endpoint_total[endpoint], 1)
            for endpoint, errors in self.errors_by_endpoint.items()
        }

    def recent_errors(self, n: int = 20) -> list[AccessLog]:
        return [
            req for req in reversed(self.recent_requests)
            if req.status_code >= 400
        ][:n]


# Sử dụng
def parse_log_stream(file_path: str) -> Iterator[AccessLog]:
    """Generator — đọc file theo từng dòng, không load toàn bộ."""
    from pathlib import Path
    with Path(file_path).open("r", encoding="utf-8") as f:
        for line in f:
            parts = line.strip().split("|")
            if len(parts) == 5:
                yield AccessLog(
                    timestamp=datetime.fromisoformat(parts[0]),
                    ip=parts[1],
                    endpoint=parts[2],
                    status_code=int(parts[3]),
                    response_time_ms=float(parts[4]),
                )

analyzer = AccessAnalyzer(recent_buffer_size=5000)
analyzer.ingest(parse_log_stream("/var/log/access.log"))

print(analyzer.top_ips(5))
print(analyzer.error_rate_by_endpoint())

Phân tích:

Counter đếm IP không cần khởi tạo thủ công — chỉ cần +=1
defaultdict(list) gom lỗi theo endpoint mà không cần if key not in dict
deque(maxlen=5000) tự loại bỏ request cũ, bộ nhớ luôn giới hạn
NamedTuple cho AccessLog — immutable, lightweight, hỗ trợ unpacking
Generator parse_log_stream đảm bảo file 10GB cũng không bị OOM

Sai lầm điển hình

❌ Sai lầm 1: defaultdict tạo key khi chỉ đọc

Vấn đề: Kiểm tra sự tồn tại của key bằng d[key] thay vì .get().

python

# SAI: kiểm tra key tạo side effect
d = defaultdict(list)
if d["missing"]:    # Tạo key "missing" với value []!
    pass
print(dict(d))      # {'missing': []}  — key phantom

Tại sao sai: Trong production, điều này gây "key phantom" — hàng ngàn key rỗng tích tụ dần, làm tăng memory và gây nhầm lẫn khi serialize ra JSON.

python

# ĐÚNG: dùng .get() hoặc toán tử in
d = defaultdict(list)
if d.get("missing"):       # Không tạo key
    pass
if "missing" in d:         # Không tạo key
    pass

❌ Sai lầm 2: Counter cho phép giá trị âm

Vấn đề: Sau subtract(), Counter chứa giá trị âm nhưng vẫn hoạt động bình thường.

python

# SAI: không kiểm tra sau subtract
inventory = Counter(apple=3, banana=1)
sold = Counter(apple=5)
inventory.subtract(sold)
print(inventory)  # Counter({'banana': 1, 'apple': -2}) — âm!

Tại sao sai: Logic kinh doanh thường không cho phép tồn kho âm. Giá trị âm lan truyền qua các phép tính tiếp theo, gây kết quả sai.

python

# ĐÚNG: dùng unary + để loại bỏ giá trị ≤ 0, hoặc kiểm tra trước
inventory = Counter(apple=3, banana=1)
sold = Counter(apple=5)
inventory.subtract(sold)
inventory = +inventory    # Counter({'banana': 1}) — loại bỏ apple

❌ Sai lầm 3: deque trong môi trường đa luồng

Vấn đề: Giả định deque thread-safe cho cặp thao tác check-then-act.

python

# SAI: check và pop không atomic
from collections import deque

d = deque()

def consumer():
    while True:
        if d:                   # Thread A kiểm tra: có phần tử
            item = d.popleft()  # Thread B đã pop trước → IndexError!

Tại sao sai: Mỗi thao tác riêng lẻ (append, popleft) trên deque là thread-safe nhờ GIL. Nhưng cặp thao tác kiểm tra-rồi-lấy thì không.

python

# ĐÚNG: dùng queue.Queue cho multi-threaded
from queue import Queue

q: Queue[int] = Queue()
q.put(42)
item = q.get()  # Thread-safe, blocking nếu rỗng

❌ Sai lầm 4: Dùng deque khi cần random access

Vấn đề: Truy cập deque[i] ở vị trí giữa — O(n).

python

# SAI: dùng deque như list
from collections import deque

d = deque(range(100_000))
value = d[50_000]  # O(n) — phải duyệt từ đầu!

Tại sao sai: Deque được triển khai dạng doubly-linked list of blocks. Truy cập index giữa phải duyệt tuần tự, không phải O(1) như list.

python

# ĐÚNG: dùng list cho random access, deque chỉ cho queue operations
data = list(range(100_000))
value = data[50_000]  # O(1)

Under the Hood

Hiệu năng

Container	Thao tác chính	Time	Space	Ghi chú
`Counter`	Khởi tạo từ iterable	O(n)	O(k)	k = số phần tử unique
`Counter`	`most_common(m)`	O(k log m)	O(m)	Dùng heapq nội bộ
`defaultdict`	`__getitem__` (key mới)	O(1)	O(1)	Gọi factory function
`deque`	`append` / `appendleft`	O(1)	O(1)	Amortized
`deque`	`__getitem__[i]`	O(n)	O(1)	Duyệt linked blocks
`OrderedDict`	`move_to_end`	O(1)	O(1)	Doubly-linked list
`ChainMap`	Lookup	O(m)	O(1)	m = số dict trong chain

Nội bộ triển khai

deque trong CPython: Không phải linked list đơn thuần. Deque dùng mảng các block cố định (mỗi block chứa 64 phần tử). Khi thêm phần tử vượt block, CPython cấp phát block mới và nối vào danh sách liên kết. Điều này giải thích tại sao append/popleft là O(1) amortized, nhưng random access phải duyệt qua các block.

Counter: Kế thừa từ dict và override __missing__ để trả về 0 cho key chưa tồn tại. Phương thức most_common() dùng heapq.nlargest() khi n < len(self), và sorted() khi lấy toàn bộ.

ChainMap: Không copy dữ liệu. Nó giữ tham chiếu đến các dict gốc trong attribute .maps. Khi lookup, nó duyệt tuần tự qua .maps và trả về kết quả đầu tiên tìm thấy. Ghi (__setitem__) chỉ ảnh hưởng dict đầu tiên.

Trade-offs

defaultdict vs setdefault(): defaultdict nhanh hơn cho pattern lặp (factory function chỉ gọi một lần, không tạo object mỗi lần như setdefault). Nhưng setdefault an toàn hơn vì không tạo key phantom khi đọc.
Counter vs manual counting: Counter tối ưu hơn nhờ triển khai C, nhưng most_common() tạo list mới mỗi lần gọi — không phù hợp trong vòng lặp tight loop.
deque vs list: Deque thắng ở hai đầu, thua ở random access và slicing. Nếu cần cả hai, xem xét sortedcontainers.SortedList hoặc cấu trúc dữ liệu chuyên biệt hơn.

Checklist ghi nhớ

✅ Checklist triển khai

Chọn container đúng

[ ] Cần đếm tần suất → dùng Counter, không viết dict thủ công
[ ] Cần nhóm phần tử theo key → dùng defaultdict(list)
[ ] Cần queue FIFO/LIFO → dùng deque, không dùng list.pop(0)
[ ] Cần immutable record có tên → dùng NamedTuple
[ ] Cần tra cứu config theo ưu tiên → dùng ChainMap

Tránh bẫy phổ biến

[ ] Không truy cập defaultdict[key] để kiểm tra — dùng .get() hoặc in
[ ] Sau Counter.subtract() — kiểm tra giá trị âm bằng +counter
[ ] Không dùng deque[i] cho random access — O(n)
[ ] deque trong multi-thread → chuyển sang queue.Queue

Production

[ ] deque(maxlen=N) cho buffer có giới hạn bộ nhớ
[ ] Counter.most_common(n) thay vì sort toàn bộ dict
[ ] ChainMap cho layered config — không merge dict thủ công

Bài tập luyện tập

Bài 1: Phân tích tần suất từ — Foundation

Đề bài: Viết hàm nhận chuỗi văn bản, trả về top N từ xuất hiện nhiều nhất (bỏ qua case). Nếu hai từ có cùng tần suất, sắp xếp theo alphabet.

python

def top_words(text: str, n: int = 5) -> list[tuple[str, int]]:
    # Viết code tại đây
    pass

# Test
text = "the cat sat on the mat the cat ate the rat"
assert top_words(text, 2) == [("the", 4), ("cat", 2)]

🧠 Quiz

Câu hỏi: Counter("aabbc").most_common() trả về gì?

[ ] A. {'a': 2, 'b': 2, 'c': 1}
[x] B. [('a', 2), ('b', 2), ('c', 1)]
[ ] C. [('c', 1), ('b', 2), ('a', 2)]
[ ] D. Counter({'a': 2, 'b': 2, 'c': 1})Giải thích: most_common() không tham số trả về list of tuples sắp xếp giảm dần theo count. Các phần tử cùng count giữ thứ tự chèn. B đúng vì trả về list, không phải dict hay Counter.

💡 Gợi ý

Dùng Counter với str.lower().split()
most_common() không đảm bảo thứ tự alphabet cho cùng tần suất — cần sort thêm

✅ Lời giải

python

from collections import Counter

def top_words(text: str, n: int = 5) -> list[tuple[str, int]]:
    counts = Counter(text.lower().split())
    sorted_items = sorted(counts.items(), key=lambda x: (-x[1], x[0]))
    return sorted_items[:n]

Phân tích: Time O(W log W) với W = số từ unique. Counter đếm O(N) với N = tổng từ. Sort thêm O(W log W) để đảm bảo thứ tự alphabet cho cùng tần suất.

Bài 2: Sliding window maximum — Intermediate

Đề bài: Cho mảng số nguyên và kích thước cửa sổ k, trả về giá trị lớn nhất trong mỗi cửa sổ trượt. Yêu cầu: O(n) time complexity.

python

def max_sliding_window(nums: list[int], k: int) -> list[int]:
    # Viết code tại đây
    pass

# Test
assert max_sliding_window([1, 3, -1, -3, 5, 3, 6, 7], 3) == [3, 3, 5, 5, 6, 7]

🧠 Quiz

Câu hỏi: Tại sao deque phù hợp cho bài toán sliding window maximum?

[ ] A. Vì deque có phương thức max() tích hợp
[ ] B. Vì deque có kích thước cố định maxlen
[x] C. Vì deque cho phép xóa phần tử ở cả hai đầu trong O(1)
[ ] D. Vì deque sắp xếp phần tử tự động Giải thích: Thuật toán monotonic deque cần xóa phần tử ở đầu (hết phạm vi cửa sổ) và cuối (nhỏ hơn phần tử mới). Cả hai thao tác đều O(1) trên deque, cho tổng O(n).

✅ Lời giải

python

from collections import deque

def max_sliding_window(nums: list[int], k: int) -> list[int]:
    dq: deque[int] = deque()  # Lưu index, giữ giá trị giảm dần
    result: list[int] = []

    for i, num in enumerate(nums):
        # Xóa phần tử ngoài cửa sổ
        while dq and dq[0] <= i - k:
            dq.popleft()

        # Giữ deque giảm dần: xóa phần tử nhỏ hơn num
        while dq and nums[dq[-1]] <= num:
            dq.pop()

        dq.append(i)

        if i >= k - 1:
            result.append(nums[dq[0]])

    return result

Phân tích: Time O(n) — mỗi phần tử vào/ra deque tối đa 1 lần. Space O(k) cho deque.

Liên kết học tiếp

Từ khóa glossary: Counter, defaultdict, deque, namedtuple, ChainMap, OrderedDict, collections module

Tìm kiếm liên quan: đếm tần suất python, hàng đợi hai đầu, dict mặc định, data pipeline python

Module collections — Cấu trúc dữ liệu chuyên biệt ​

Bức tranh tư duy ​

Cốt lõi kỹ thuật ​

Counter — Bảng tần suất ​

defaultdict — Dict tự khởi tạo ​

OrderedDict — Dict giữ thứ tự chèn ​

deque — Hàng đợi hai đầu ​

namedtuple — Tuple có tên trường ​

ChainMap — Tra cứu theo chuỗi ưu tiên ​

Thực chiến ​

Tình huống: Pipeline xử lý log truy cập web server ​

Sai lầm điển hình ​

❌ Sai lầm 1: defaultdict tạo key khi chỉ đọc ​

❌ Sai lầm 2: Counter cho phép giá trị âm ​

❌ Sai lầm 3: deque trong môi trường đa luồng ​

❌ Sai lầm 4: Dùng deque khi cần random access ​

Under the Hood ​

Hiệu năng ​

Nội bộ triển khai ​

Trade-offs ​

Checklist ghi nhớ ​

Bài tập luyện tập ​

Bài 1: Phân tích tần suất từ — Foundation ​

Bài 2: Sliding window maximum — Intermediate ​

Liên kết học tiếp ​

Module collections — Cấu trúc dữ liệu chuyên biệt

Bức tranh tư duy

Cốt lõi kỹ thuật

Counter — Bảng tần suất

defaultdict — Dict tự khởi tạo

OrderedDict — Dict giữ thứ tự chèn

deque — Hàng đợi hai đầu

namedtuple — Tuple có tên trường

ChainMap — Tra cứu theo chuỗi ưu tiên

Thực chiến

Tình huống: Pipeline xử lý log truy cập web server

Sai lầm điển hình

❌ Sai lầm 1: defaultdict tạo key khi chỉ đọc

❌ Sai lầm 2: Counter cho phép giá trị âm

❌ Sai lầm 3: deque trong môi trường đa luồng

❌ Sai lầm 4: Dùng deque khi cần random access

Under the Hood

Hiệu năng

Nội bộ triển khai

Trade-offs

Checklist ghi nhớ

Bài tập luyện tập

Bài 1: Phân tích tần suất từ — Foundation

Bài 2: Sliding window maximum — Intermediate

Liên kết học tiếp