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 にフィールド名を与えます。さらに OrderedDict と ChainMap は、通常の 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']| Class | Purpose | Replaces |
|---|---|---|
Counter | ハッシュ可能なオブジェクトを数える | dict を使った手動カウントループ |
defaultdict | 自動デフォルト値を持つ dict | dict.setdefault()、if key not in チェック |
deque | 両端キュー(末端操作が O(1)) | キュー/スタック用途の list |
namedtuple | 名前付きフィールドを持つ tuple | 素の tuple、簡易データクラス |
OrderedDict | 挿入順を覚える dict | dict(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")) # FalseCounter をさらに深掘りしたい場合は、専用の 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| Operation | list | deque |
|---|---|---|
append(x) (right) | O(1) amortized | O(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 の比較
| Feature | namedtuple | dataclass |
|---|---|---|
| デフォルトで immutable | Yes | No(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)すべてのコレクション型の比較
| Type | Base Class | Mutable | Use Case | Key Advantage |
|---|---|---|---|---|
Counter | dict | Yes | 要素カウント | most_common()、多重集合演算 |
defaultdict | dict | Yes | 欠損キーの自動初期化 | KeyError 回避、factory function |
deque | -- | Yes | 両端キュー | 両端 O(1)、maxlen |
namedtuple | tuple | No | 構造化データレコード | フィールド名アクセス、軽量 |
OrderedDict | dict | Yes | 順序が重要な dict | move_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 によるコレクションデータの可視化
Counter、defaultdict、namedtuple でデータを処理した後は、結果を可視化したくなることがよくあります。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 entriesFAQ
Python の collections モジュールとは?
collections モジュールは Python 標準ライブラリの一部です。組み込み型(dict / list / tuple / set)を拡張する特殊なコンテナデータ型を提供します。主なクラスは Counter、defaultdict、deque、namedtuple、OrderedDict、ChainMap です。どれも、組み込み型だけでは冗長になりがちな特定カテゴリのデータ処理を、より効率よく扱うために設計されています。
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 サブクラスを作ります。軽量で、反復・アンパックができ、メモリ使用量も最小です。dataclass(dataclasses モジュール、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 は手書きのカウントループを置き換え、defaultdict は KeyError 処理を不要にし、deque は高速な両端操作を実現します。namedtuple は tuple に読みやすいフィールド名を与え、OrderedDict は順序に敏感な比較と再順序化を扱い、ChainMap はデータを複製せずにレイヤー化された dict 参照を管理します。
それぞれが、組み込みコンテナよりもうまく解ける「特定の問題」を持っています。重要なのは、操作パターンに合わせてデータ構造を選ぶことです。カウント(Counter)、グルーピング(defaultdict)、キュー/スタック(deque)、構造化レコード(namedtuple)、順序操作(OrderedDict)、レイヤー検索(ChainMap)。これを意識するだけで、Python コードは短く、速く、保守しやすくなります。