Skip to content

Python collections モジュール解説: Counter / defaultdict / deque / namedtuple ガイド

Updated on

Python の組み込みデータ構造(list / dict / tuple / set)だけで、たいていのタスクはこなせます。ですが、コードが「おもちゃの例」を超えて大きくなると、少しずつ限界が見えてきます。要素のカウントは辞書の手作業ループが必要になり、データのグルーピングでは if key not in dict のチェックが散らばります。list をキューとして使うと先頭からの pop が O(n) になって性能が落ちます。構造化レコードをただの tuple で表現すると、フィールド参照が「何番目の要素だっけ?」という読みづらい当て推量ゲームになりがちです。どれも小さな回避策ですが、積み重なると可読性を下げ、実行を遅くし、壊れやすいコードになっていきます。

Python 標準ライブラリの collections モジュールは、こうした問題を目的別に設計されたコンテナ型で解決します。Counter は一発で要素を数え、defaultdict は自動デフォルト値で KeyError を防ぎ、deque は両端の操作を O(1) にし、namedtuple はクラスを作るほどではない場面で tuple にフィールド名を与えます。さらに OrderedDictChainMap は、通常の dict では表現しにくい「順序」や「レイヤー化された検索」をきれいに扱えます。

このガイドでは、collections モジュールの主要クラスをすべて、動くコード、性能面の説明、実務パターンとともに解説します。ログ処理、キャッシュの構築、設定のレイヤリング、データパイプラインの構造化など、さまざまな場面でこれらのコンテナはコードを短く・速く・正しくしてくれます。

📚

collections モジュール概要

collections モジュールは、Python の汎用的な組み込みコンテナを拡張する、特殊なコンテナデータ型を提供します。

import collections
 
# See all available classes
print([name for name in dir(collections) if not name.startswith('_')])
# ['ChainMap', 'Counter', 'OrderedDict', 'UserDict', 'UserList',
#  'UserString', 'abc', 'defaultdict', 'deque', 'namedtuple']
ClassPurposeReplaces
Counterハッシュ可能なオブジェクトを数えるdict を使った手動カウントループ
defaultdict自動デフォルト値を持つ dictdict.setdefault()if key not in チェック
deque両端キュー(末端操作が O(1))キュー/スタック用途の list
namedtuple名前付きフィールドを持つ tuple素の tuple、簡易データクラス
OrderedDict挿入順を覚える dictdict(3.7 より前)、順序を使う操作
ChainMapレイヤー化された dict 参照手動での dict マージ

Counter: 要素を数える

Counter は、ハッシュ可能なオブジェクトを数えるための dict サブクラスです。要素をキー、出現回数を値として保持し、頻度分析向けのメソッドを提供します。

Counter の作成

from collections import Counter
 
# From an iterable
words = ['apple', 'banana', 'apple', 'cherry', 'banana', 'apple']
word_count = Counter(words)
print(word_count)
# Counter({'apple': 3, 'banana': 2, 'cherry': 1})
 
# From a string
letter_count = Counter('mississippi')
print(letter_count)
# Counter({'s': 4, 'i': 4, 'p': 2, 'm': 1})
 
# From a dictionary
inventory = Counter({'shirts': 25, 'pants': 15, 'hats': 10})
 
# From keyword arguments
stock = Counter(laptops=5, monitors=12)

most_common() と頻度ランキング

from collections import Counter
 
text = "to be or not to be that is the question"
words = Counter(text.split())
 
# Get the 3 most common words
print(words.most_common(3))
# [('to', 2), ('be', 2), ('or', 1)]
 
# Get all elements sorted by frequency
print(words.most_common())
# [('to', 2), ('be', 2), ('or', 1), ('not', 1), ('that', 1), ('is', 1), ('the', 1), ('question', 1)]
 
# Least common: reverse the list or slice from the end
print(words.most_common()[-3:])
# [('is', 1), ('the', 1), ('question', 1)]

Counter の算術演算

Counter は、多重集合(multiset)として扱うことで、加算・減算・積(共通部分)・和(最大)などの演算をサポートします。

from collections import Counter
 
a = Counter(x=4, y=2, z=1)
b = Counter(x=1, y=3, z=5)
 
# Addition: combine counts
print(a + b)  # Counter({'z': 6, 'y': 5, 'x': 5})
 
# Subtraction: drops zero and negative results
print(a - b)  # Counter({'x': 3})
 
# Intersection (min of each)
print(a & b)  # Counter({'y': 2, 'x': 1, 'z': 1})
 
# Union (max of each)
print(a | b)  # Counter({'z': 5, 'x': 4, 'y': 3})

Counter の実用パターン

from collections import Counter
 
# Word frequency analysis
log_entries = [
    "ERROR: disk full",
    "WARNING: high memory",
    "ERROR: disk full",
    "ERROR: timeout",
    "WARNING: high memory",
    "ERROR: disk full",
    "INFO: backup complete",
]
error_types = Counter(entry.split(":")[0].strip() for entry in log_entries)
print(error_types)
# Counter({'ERROR': 4, 'WARNING': 2, 'INFO': 1})
 
# Find unique elements (count == 1)
data = [1, 2, 3, 2, 1, 4, 5, 4]
unique = [item for item, count in Counter(data).items() if count == 1]
print(unique)  # [3, 5]
 
# Check if one collection is a subset of another (anagram check)
def is_anagram(word1, word2):
    return Counter(word1.lower()) == Counter(word2.lower())
 
print(is_anagram("listen", "silent"))  # True
print(is_anagram("hello", "world"))    # False

Counter をさらに深掘りしたい場合は、専用の Python Counter guide を参照してください。

defaultdict: 自動デフォルト値

defaultdict は dict のサブクラスで、存在しないキーにアクセスしたときに factory function を呼び出してデフォルト値を供給します。これにより KeyError を回避でき、防御的なチェックが不要になります。

Factory function

from collections import defaultdict
 
# int factory: default is 0
counter = defaultdict(int)
counter['apples'] += 1
counter['oranges'] += 3
print(dict(counter))  # {'apples': 1, 'oranges': 3}
 
# list factory: default is []
groups = defaultdict(list)
pairs = [('fruit', 'apple'), ('veggie', 'carrot'), ('fruit', 'banana'), ('veggie', 'pea')]
for category, item in pairs:
    groups[category].append(item)
print(dict(groups))
# {'fruit': ['apple', 'banana'], 'veggie': ['carrot', 'pea']}
 
# set factory: default is set()
index = defaultdict(set)
words = [('file1', 'python'), ('file2', 'python'), ('file1', 'java'), ('file3', 'python')]
for filename, lang in words:
    index[lang].add(filename)
print(dict(index))
# {'python': {'file1', 'file2', 'file3'}, 'java': {'file1'}}

グルーピングの定番パターン

関連データのグルーピングは defaultdict(list) の最頻出用途です。手動のやり方と比較すると違いが明確です。

from collections import defaultdict
 
students = [
    ('Math', 'Alice'), ('Science', 'Bob'), ('Math', 'Charlie'),
    ('Science', 'Diana'), ('Math', 'Eve'), ('History', 'Frank'),
]
 
# Without defaultdict -- verbose and error-prone
groups_manual = {}
for subject, name in students:
    if subject not in groups_manual:
        groups_manual[subject] = []
    groups_manual[subject].append(name)
 
# With defaultdict -- clean and direct
groups = defaultdict(list)
for subject, name in students:
    groups[subject].append(name)
 
print(dict(groups))
# {'Math': ['Alice', 'Charlie', 'Eve'], 'Science': ['Bob', 'Diana'], 'History': ['Frank']}

ネストした defaultdict

各レベルを手動で初期化せずに、多段のデータ構造を構築できます。

from collections import defaultdict
 
# Two-level nested defaultdict
def nested_dict():
    return defaultdict(int)
 
sales = defaultdict(nested_dict)
sales['2025']['Q1'] = 150000
sales['2025']['Q2'] = 175000
sales['2026']['Q1'] = 200000
print(sales['2025']['Q1'])  # 150000
print(sales['2024']['Q3'])  # 0 (auto-created, no KeyError)
 
# Arbitrary depth nesting with a recursive factory
def deep_dict():
    return defaultdict(deep_dict)
 
config = deep_dict()
config['database']['primary']['host'] = 'localhost'
config['database']['primary']['port'] = 5432
config['database']['replica']['host'] = 'replica.local'
print(config['database']['primary']['host'])  # localhost

カスタム factory function

from collections import defaultdict
 
# Lambda for custom defaults
scores = defaultdict(lambda: 100)  # Every student starts with 100
scores['Alice'] -= 5
scores['Bob'] -= 10
print(scores['Charlie'])  # 100 (new student gets default)
print(dict(scores))  # {'Alice': 95, 'Bob': 90, 'Charlie': 100}
 
# Named function for complex defaults
def default_user():
    return {'role': 'viewer', 'active': True, 'login_count': 0}
 
users = defaultdict(default_user)
users['alice']['role'] = 'admin'
print(users['bob'])  # {'role': 'viewer', 'active': True, 'login_count': 0}

さらにパターンを知りたい場合は Python defaultdict guide を参照してください。

deque: 両端キュー

deque(発音は “deck”)は、両端での append / pop を O(1) で提供します。list は pop(0)insert(0, x) が O(n) になり、全要素のシフトが発生します。シーケンスの両端に触るワークロードでは、deque を使うのが正解です。

基本操作

from collections import deque
 
d = deque([1, 2, 3, 4, 5])
 
# O(1) operations on both ends
d.append(6)         # Add to right: [1, 2, 3, 4, 5, 6]
d.appendleft(0)     # Add to left:  [0, 1, 2, 3, 4, 5, 6]
 
right = d.pop()     # Remove from right: 6
left = d.popleft()  # Remove from left:  0
print(d)  # deque([1, 2, 3, 4, 5])
 
# Extend from both sides
d.extend([6, 7])          # Right extend: [1, 2, 3, 4, 5, 6, 7]
d.extendleft([-1, 0])     # Left extend (reversed): [0, -1, 1, 2, 3, 4, 5, 6, 7]

maxlen 付きの bounded deque

maxlen を設定すると、上限を超えて要素を追加した際に反対側の要素が自動的に捨てられます。スライディングウィンドウやキャッシュに最適です。

from collections import deque
 
# Keep only the last 5 items
recent = deque(maxlen=5)
for i in range(10):
    recent.append(i)
 
print(recent)  # deque([5, 6, 7, 8, 9], maxlen=5)
 
# Sliding window average
def moving_average(iterable, window_size):
    window = deque(maxlen=window_size)
    for value in iterable:
        window.append(value)
        if len(window) == window_size:
            yield sum(window) / window_size
 
data = [10, 20, 30, 40, 50, 60, 70]
print(list(moving_average(data, 3)))
# [20.0, 30.0, 40.0, 50.0, 60.0]

回転(Rotation)

rotate(n) は要素を右に n ステップずらします。負の値は左回転です。

from collections import deque
 
d = deque([1, 2, 3, 4, 5])
 
d.rotate(2)   # Rotate right by 2
print(d)  # deque([4, 5, 1, 2, 3])
 
d.rotate(-3)  # Rotate left by 3
print(d)  # deque([2, 3, 4, 5, 1])

deque と list の性能差

from collections import deque
import time
 
# Benchmark: append/pop from left side
n = 100_000
 
# List: O(n) for each insert at position 0
start = time.perf_counter()
lst = []
for i in range(n):
    lst.insert(0, i)
list_time = time.perf_counter() - start
 
# Deque: O(1) for appendleft
start = time.perf_counter()
dq = deque()
for i in range(n):
    dq.appendleft(i)
deque_time = time.perf_counter() - start
 
print(f"List insert(0, x): {list_time:.4f}s")
print(f"Deque appendleft:  {deque_time:.4f}s")
print(f"Deque is {list_time / deque_time:.0f}x faster")
# Typical output:
# List insert(0, x): 1.2340s
# Deque appendleft:  0.0065s
# Deque is 190x faster
Operationlistdeque
append(x) (right)O(1) amortizedO(1)
pop() (right)O(1)O(1)
insert(0, x) / appendleft(x)O(n)O(1)
pop(0) / popleft()O(n)O(1)
access by index [i]O(1)O(n)
Memory per element低いやや高い

両端で高速に操作したいなら deque。インデックスによるランダムアクセスが重要なら list を使いましょう。

完全版は Python deque を参照してください。

namedtuple: 名前付きフィールドを持つ tuple

namedtuple は、名前付きフィールドを持つ tuple のサブクラスを生成します。フルクラス定義のオーバーヘッドなしに、コードを自己記述的にできます。

namedtuple の作成

from collections import namedtuple
 
# Define a type
Point = namedtuple('Point', ['x', 'y'])
p = Point(3, 4)
 
# Access by name or index
print(p.x)     # 3
print(p[1])    # 4
print(p)       # Point(x=3, y=4)
 
# Alternative field definition styles
Color = namedtuple('Color', 'red green blue')        # Space-separated string
Config = namedtuple('Config', 'host, port, database')  # Comma-separated string

なぜ素の tuple ではなく namedtuple を使うのか?

from collections import namedtuple
 
# Plain tuple: which index is what?
employee_tuple = ('Alice', 'Engineering', 95000, True)
print(employee_tuple[2])  # 95000 -- but what does index 2 mean?
 
# namedtuple: self-documenting
Employee = namedtuple('Employee', 'name department salary active')
employee = Employee('Alice', 'Engineering', 95000, True)
print(employee.salary)     # 95000 -- immediately clear
print(employee.department) # Engineering

主要メソッド

from collections import namedtuple
 
Employee = namedtuple('Employee', 'name department salary')
emp = Employee('Alice', 'Engineering', 95000)
 
# _replace: create a new instance with some fields changed (immutable)
promoted = emp._replace(salary=110000)
print(promoted)  # Employee(name='Alice', department='Engineering', salary=110000)
print(emp)       # Employee(name='Alice', department='Engineering', salary=95000)  -- unchanged
 
# _asdict: convert to OrderedDict (Python 3.8+ returns regular dict)
print(emp._asdict())
# {'name': 'Alice', 'department': 'Engineering', 'salary': 95000}
 
# _fields: get field names
print(Employee._fields)  # ('name', 'department', 'salary')
 
# _make: create from an iterable
data = ['Bob', 'Marketing', 85000]
emp2 = Employee._make(data)
print(emp2)  # Employee(name='Bob', department='Marketing', salary=85000)

デフォルト値

from collections import namedtuple
 
# defaults parameter (Python 3.6.1+)
Connection = namedtuple('Connection', 'host port timeout', defaults=[5432, 30])
conn1 = Connection('localhost')               # port=5432, timeout=30
conn2 = Connection('db.example.com', 3306)    # timeout=30
conn3 = Connection('db.example.com', 3306, 60)
 
print(conn1)  # Connection(host='localhost', port=5432, timeout=30)
print(conn2)  # Connection(host='db.example.com', port=3306, timeout=30)

typing.NamedTuple という選択肢

型注釈や、よりクラスらしい構文が欲しい場合は typing.NamedTuple を使えます。

from typing import NamedTuple
 
class Point(NamedTuple):
    x: float
    y: float
    label: str = "origin"
 
p = Point(3.0, 4.0, "A")
print(p.x, p.label)  # 3.0 A
 
# Still a tuple -- supports unpacking, indexing, iteration
x, y, label = p
print(f"({x}, {y})")  # (3.0, 4.0)

namedtuple と dataclass の比較

Featurenamedtupledataclass
デフォルトで immutableYesNo(frozen=True が必要)
メモリフットプリントtuple と同等(小さい)大きい(通常のクラス)
反復/アンパックYes(tuple だから)No(メソッド追加が必要)
型注釈typing.NamedTuple で対応標準対応
メソッド/プロパティサブクラス化が必要直接サポート
継承制限あり通常のクラス継承
最適用途軽量なデータレコード複雑で mutable なオブジェクト

OrderedDict: 順序付き dict 操作

Python 3.7 以降は通常の dict も挿入順を保持します。それでも OrderedDict が必要になるのはいつでしょうか?

OrderedDict がまだ重要なケース

from collections import OrderedDict
 
# 1. Equality considers order
d1 = {'a': 1, 'b': 2}
d2 = {'b': 2, 'a': 1}
print(d1 == d2)  # True -- regular dicts ignore order in comparison
 
od1 = OrderedDict([('a', 1), ('b', 2)])
od2 = OrderedDict([('b', 2), ('a', 1)])
print(od1 == od2)  # False -- OrderedDict considers order
 
# 2. move_to_end() for reordering
od = OrderedDict([('a', 1), ('b', 2), ('c', 3)])
od.move_to_end('a')           # Move 'a' to the end
print(list(od.keys()))  # ['b', 'c', 'a']
 
od.move_to_end('c', last=False)  # Move 'c' to the beginning
print(list(od.keys()))  # ['c', 'b', 'a']

OrderedDict で LRU キャッシュを作る

from collections import OrderedDict
 
class LRUCache:
    def __init__(self, capacity):
        self.cache = OrderedDict()
        self.capacity = capacity
 
    def get(self, key):
        if key not in self.cache:
            return -1
        self.cache.move_to_end(key)  # Mark as recently used
        return self.cache[key]
 
    def put(self, key, value):
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = value
        if len(self.cache) > self.capacity:
            self.cache.popitem(last=False)  # Remove oldest
 
cache = LRUCache(3)
cache.put('a', 1)
cache.put('b', 2)
cache.put('c', 3)
cache.get('a')       # Access 'a', moves it to end
cache.put('d', 4)    # Evicts 'b' (least recently used)
print(list(cache.cache.keys()))  # ['c', 'a', 'd']

ChainMap: レイヤー化された dict 参照

ChainMap は複数の dict を単一ビューとして扱い、参照時に順番に検索します。最初に見つかった値を返すため、設定のレイヤリング、スコープ付き変数参照、コンテキスト管理に向いています。

基本的な使い方

from collections import ChainMap
 
defaults = {'theme': 'light', 'language': 'en', 'timeout': 30}
user_prefs = {'theme': 'dark'}
session = {'language': 'fr'}
 
config = ChainMap(session, user_prefs, defaults)
 
# Lookup searches session -> user_prefs -> defaults
print(config['theme'])     # 'dark'    (from user_prefs)
print(config['language'])  # 'fr'      (from session)
print(config['timeout'])   # 30        (from defaults)

設定のレイヤリング

from collections import ChainMap
import os
 
# Real-world config pattern: CLI args > env vars > config file > defaults
defaults = {
    'debug': False,
    'log_level': 'WARNING',
    'port': 8080,
    'host': '0.0.0.0',
}
 
config_file = {
    'log_level': 'INFO',
    'port': 9090,
}
 
env_vars = {
    k.lower(): v for k, v in os.environ.items()
    if k.lower() in defaults
}
 
cli_args = {'debug': True}  # Parsed from argparse
 
config = ChainMap(cli_args, env_vars, config_file, defaults)
print(config['debug'])      # True (from cli_args)
print(config['log_level'])  # 'INFO' (from config_file)
print(config['host'])       # '0.0.0.0' (from defaults)

new_child() によるスコープ(文脈)管理

from collections import ChainMap
 
# Simulating variable scoping (like nested function scopes)
global_scope = {'x': 1, 'y': 2}
local_scope = ChainMap(global_scope)
 
# Enter a new scope
inner_scope = local_scope.new_child()
inner_scope['x'] = 10  # Shadows global x
inner_scope['z'] = 30  # New local variable
 
print(inner_scope['x'])  # 10 (local)
print(inner_scope['y'])  # 2  (falls through to global)
print(inner_scope['z'])  # 30 (local)
 
# Exit scope -- original is unchanged
print(local_scope['x'])  # 1 (global still intact)

すべてのコレクション型の比較

TypeBase ClassMutableUse CaseKey Advantage
CounterdictYes要素カウントmost_common()、多重集合演算
defaultdictdictYes欠損キーの自動初期化KeyError 回避、factory function
deque--Yes両端キュー両端 O(1)、maxlen
namedtupletupleNo構造化データレコードフィールド名アクセス、軽量
OrderedDictdictYes順序が重要な dictmove_to_end()、順序を考慮した等価比較
ChainMap--Yesレイヤー検索設定レイヤリング、スコープ

パフォーマンスベンチマーク

Counter vs 手動カウント

from collections import Counter, defaultdict
import time
 
data = list(range(1000)) * 1000  # 1 million items, 1000 unique
 
# Method 1: Counter
start = time.perf_counter()
c = Counter(data)
counter_time = time.perf_counter() - start
 
# Method 2: defaultdict(int)
start = time.perf_counter()
dd = defaultdict(int)
for item in data:
    dd[item] += 1
dd_time = time.perf_counter() - start
 
# Method 3: Manual dict
start = time.perf_counter()
manual = {}
for item in data:
    manual[item] = manual.get(item, 0) + 1
manual_time = time.perf_counter() - start
 
print(f"Counter:         {counter_time:.4f}s")
print(f"defaultdict(int):{dd_time:.4f}s")
print(f"dict.get():      {manual_time:.4f}s")
# Typical: Counter ~0.03s, defaultdict ~0.07s, dict.get() ~0.09s

キュー操作における deque vs list

from collections import deque
import time
 
n = 100_000
 
# Simulate a FIFO queue: append right, pop left
# List
start = time.perf_counter()
q = list(range(n))
while q:
    q.pop(0)
list_queue_time = time.perf_counter() - start
 
# Deque
start = time.perf_counter()
q = deque(range(n))
while q:
    q.popleft()
deque_queue_time = time.perf_counter() - start
 
print(f"List pop(0):     {list_queue_time:.4f}s")
print(f"Deque popleft(): {deque_queue_time:.4f}s")
print(f"Deque is {list_queue_time / deque_queue_time:.0f}x faster")
# Typical: List ~2.5s, Deque ~0.004s -> ~600x faster

実務的な例

Counter でログ分析

from collections import Counter
from datetime import datetime
 
# Parse and analyze server logs
log_lines = [
    "2026-02-18 10:15:03 GET /api/users 200",
    "2026-02-18 10:15:04 POST /api/login 401",
    "2026-02-18 10:15:05 GET /api/users 200",
    "2026-02-18 10:15:06 GET /api/products 500",
    "2026-02-18 10:15:07 POST /api/login 200",
    "2026-02-18 10:15:08 GET /api/users 200",
    "2026-02-18 10:15:09 GET /api/products 500",
    "2026-02-18 10:15:10 POST /api/login 401",
]
 
# Count status codes
status_codes = Counter(line.split()[-1] for line in log_lines)
print("Status codes:", status_codes.most_common())
# [('200', 4), ('401', 2), ('500', 2)]
 
# Count endpoints
endpoints = Counter(line.split()[3] for line in log_lines)
print("Top endpoints:", endpoints.most_common(2))
# [('/api/users', 3), ('/api/login', 3)]
 
# Count error endpoints (status >= 400)
errors = Counter(
    line.split()[3] for line in log_lines
    if int(line.split()[-1]) >= 400
)
print("Error endpoints:", errors)
# Counter({'/api/login': 2, '/api/products': 2})

ChainMap による設定管理

from collections import ChainMap
import json
 
# Multi-layer config system for a web application
def load_config(config_path=None, cli_overrides=None):
    # Layer 1: Hard-coded defaults
    defaults = {
        'host': '127.0.0.1',
        'port': 8000,
        'debug': False,
        'db_pool_size': 5,
        'log_level': 'WARNING',
        'cors_origins': ['http://localhost:3000'],
    }
 
    # Layer 2: Config file
    file_config = {}
    if config_path:
        with open(config_path) as f:
            file_config = json.load(f)
 
    # Layer 3: CLI overrides (highest priority)
    cli = cli_overrides or {}
 
    # ChainMap searches cli -> file_config -> defaults
    return ChainMap(cli, file_config, defaults)
 
# Usage
config = load_config(cli_overrides={'debug': True, 'port': 9000})
print(config['debug'])        # True (CLI override)
print(config['port'])         # 9000 (CLI override)
print(config['db_pool_size']) # 5    (default)
print(config['log_level'])    # WARNING (default)

deque による最近アイテムキャッシュ

from collections import deque
 
class RecentItemsTracker:
    """Track the N most recent unique items."""
 
    def __init__(self, max_items=10):
        self.items = deque(maxlen=max_items)
        self.seen = set()
 
    def add(self, item):
        if item in self.seen:
            # Move to front by removing and re-adding
            self.items.remove(item)
            self.items.append(item)
        else:
            if len(self.items) == self.items.maxlen:
                # Remove the oldest item from the set too
                oldest = self.items[0]
                self.seen.discard(oldest)
            self.items.append(item)
            self.seen.add(item)
 
    def get_recent(self):
        return list(reversed(self.items))
 
# Track recently viewed products
tracker = RecentItemsTracker(max_items=5)
for product in ['shoes', 'shirt', 'hat', 'shoes', 'jacket', 'belt', 'hat']:
    tracker.add(product)
 
print(tracker.get_recent())
# ['hat', 'belt', 'jacket', 'shoes', 'shirt']

namedtuple を使ったデータパイプライン

from collections import namedtuple, Counter, defaultdict
 
# Define structured records
Transaction = namedtuple('Transaction', 'id customer product amount date')
 
transactions = [
    Transaction(1, 'Alice', 'Widget', 29.99, '2026-02-01'),
    Transaction(2, 'Bob', 'Gadget', 49.99, '2026-02-01'),
    Transaction(3, 'Alice', 'Widget', 29.99, '2026-02-03'),
    Transaction(4, 'Charlie', 'Gadget', 49.99, '2026-02-05'),
    Transaction(5, 'Alice', 'Gizmo', 19.99, '2026-02-07'),
    Transaction(6, 'Bob', 'Widget', 29.99, '2026-02-08'),
]
 
# Most popular products
product_count = Counter(t.product for t in transactions)
print("Popular products:", product_count.most_common())
# [('Widget', 3), ('Gadget', 2), ('Gizmo', 1)]
 
# Revenue by customer
revenue = defaultdict(float)
for t in transactions:
    revenue[t.customer] += t.amount
print("Revenue:", dict(revenue))
# {'Alice': 79.97, 'Bob': 79.98, 'Charlie': 49.99}
 
# Convert to DataFrame for visualization
import pandas as pd
df = pd.DataFrame(transactions, columns=Transaction._fields)
print(df.groupby('customer')['amount'].sum())

PyGWalker によるコレクションデータの可視化

Counterdefaultdictnamedtuple でデータを処理した後は、結果を可視化したくなることがよくあります。PyGWalker (opens in a new tab) は、任意の pandas DataFrame を Jupyter notebooks 上で Tableau 風のインタラクティブ可視化 UI に変換できます。

from collections import Counter
import pandas as pd
import pygwalker as pyg
 
# Process data with collections
log_data = ["ERROR", "WARNING", "ERROR", "INFO", "ERROR", "WARNING", "INFO", "INFO"]
counts = Counter(log_data)
 
# Convert to DataFrame
df = pd.DataFrame(counts.items(), columns=['Level', 'Count'])
 
# Launch interactive visualization
walker = pyg.walk(df)

これにより、フィールドをドラッグ&ドロップしてチャートを作成したり、フィルタリングしたり、パターンをインタラクティブに探索できます(可視化コードを書く必要はありません)。特に Counter の集計や defaultdict のグルーピングで処理した大規模データの分布を素早く探索したいときに有用です。

これらのコレクション実験をインタラクティブに実行するなら、RunCell (opens in a new tab) は AI 支援の Jupyter 環境を提供し、即時フィードバックを得ながらデータ処理パイプラインを反復できます。

複数のコレクション型を組み合わせる

collections の真価は、複数の型を 1 本のパイプラインで組み合わせたときに発揮されます。

from collections import Counter, defaultdict, namedtuple, deque
 
# Named record type
LogEntry = namedtuple('LogEntry', 'timestamp level message')
 
# Simulated log stream
log_stream = deque([
    LogEntry('10:01', 'ERROR', 'Connection timeout'),
    LogEntry('10:02', 'INFO', 'Request processed'),
    LogEntry('10:03', 'ERROR', 'Connection timeout'),
    LogEntry('10:04', 'WARNING', 'High memory'),
    LogEntry('10:05', 'ERROR', 'Disk full'),
    LogEntry('10:06', 'INFO', 'Request processed'),
    LogEntry('10:07', 'ERROR', 'Connection timeout'),
], maxlen=100)
 
# Count error types
error_counts = Counter(
    entry.message for entry in log_stream if entry.level == 'ERROR'
)
print("Error types:", error_counts.most_common())
# [('Connection timeout', 3), ('Disk full', 1)]
 
# Group entries by level
by_level = defaultdict(list)
for entry in log_stream:
    by_level[entry.level].append(entry)
 
for level, entries in by_level.items():
    print(f"{level}: {len(entries)} entries")
# ERROR: 4 entries
# INFO: 2 entries
# WARNING: 1 entries

FAQ

Python の collections モジュールとは?

collections モジュールは Python 標準ライブラリの一部です。組み込み型(dict / list / tuple / set)を拡張する特殊なコンテナデータ型を提供します。主なクラスは CounterdefaultdictdequenamedtupleOrderedDictChainMap です。どれも、組み込み型だけでは冗長になりがちな特定カテゴリのデータ処理を、より効率よく扱うために設計されています。

Counter と defaultdict(int) はどう使い分ける?

主目的が「要素のカウント」や「頻度分布の比較」である場合は Counter を使います。most_common()、算術演算子(+ / - / & / |)が使え、イテラブル全体をコンストラクタ 1 回で数えられます。一方 defaultdict(int) は、カウントが主役ではなく、より一般的な「整数デフォルト付きの辞書」が必要なときや、別のデータ構造パターンの一部として数えるときに向きます。

Python の deque は thread-safe?

はい。CPython では GIL(Global Interpreter Lock)の影響で deque.append()deque.appendleft()deque.pop()deque.popleft() はアトミック操作です。そのため追加のロックなしに thread-safe なキューとして利用できます。ただし、複数の操作を組み合わせた「チェックしてから実行する」といった複合操作はアトミックではないため、必要に応じて明示的な同期が必要です。

namedtuple と dataclass の違いは?

namedtuple は、名前付きフィールドを持つ immutable な tuple サブクラスを作ります。軽量で、反復・アンパックができ、メモリ使用量も最小です。dataclassdataclasses モジュール、Python 3.7+)は、デフォルトで mutable な属性を持つ通常のクラスを生成し、メソッド/プロパティ/継承などを自然に扱えます。単純な immutable レコードなら namedtuple、可変性や複雑な振る舞いが必要なら dataclass を選びます。

Python 3.7+ でも OrderedDict は必要?

はい、主に 2 つのケースで必要です。1 つ目は、OrderedDict の等価比較が要素順序を考慮する点(OrderedDict(a=1, b=2) != OrderedDict(b=2, a=1))で、通常の dict は順序を無視して比較します。2 つ目は、OrderedDict が再順序化のための move_to_end() を提供する点で、LRU キャッシュなどの実装に便利です。それ以外の用途では、通常の dict で十分で、一般的により高性能です。

ChainMap は dict をマージするのと何が違う?

ChainMap は複数の dict の上に「ビュー」を作り、データをコピーしません。検索は先頭から順に行われ、下層 dict の変更も即座に ChainMap に反映されます。対して {**d1, **d2}d1 | d2 によるマージは新しい dict を作り、全データを複製します。大きな dict では ChainMap の方がメモリ効率が高く、設定のレイヤー構造も保てます。

collections の型は type hints と一緒に使える?

はい。collections.Counter[str]collections.defaultdict[str, list[int]]collections.deque[int] のように型付けできます。namedtuple については、型注釈を直接クラス定義に書ける typing.NamedTuple の利用がおすすめです。これらの型は mypy などの型チェッカーとも互換性があります。

まとめ

Python の collections モジュールは、よくあるボイラープレートを減らす 6 つの特殊なコンテナ型を提供します。Counter は手書きのカウントループを置き換え、defaultdictKeyError 処理を不要にし、deque は高速な両端操作を実現します。namedtuple は tuple に読みやすいフィールド名を与え、OrderedDict は順序に敏感な比較と再順序化を扱い、ChainMap はデータを複製せずにレイヤー化された dict 参照を管理します。

それぞれが、組み込みコンテナよりもうまく解ける「特定の問題」を持っています。重要なのは、操作パターンに合わせてデータ構造を選ぶことです。カウント(Counter)、グルーピング(defaultdict)、キュー/スタック(deque)、構造化レコード(namedtuple)、順序操作(OrderedDict)、レイヤー検索(ChainMap)。これを意識するだけで、Python コードは短く、速く、保守しやすくなります。

📚