7강: 고급 데이터 구조 및 collections 모듈 - 파이썬 고급편

 

7강: 파이썬 표준 라이브러리 속 숨은 보석 - collections 모듈 활용

개요 및 중요성

파이썬의 기본 데이터 구조(list, dict, set, tuple)만으로도 많은 작업을 수행할 수 있지만, collections 모듈은 더욱 특화되고 효율적인 데이터 구조들을 제공합니다. 이러한 고급 데이터 구조들은 특정 상황에서 놀라운 성능 향상과 코드 간결성을 제공합니다.

핵심 포인트: 올바른 데이터 구조 선택이 알고리즘 성능을 크게 좌우합니다.

Counter: 카운팅의 강자

Counter는 해시 가능한 객체를 카운팅하는 dict의 서브클래스입니다. 데이터 분석이나 통계 작업에서 매우 유용합니다.

from collections import Counter

# 기본 카운팅
text = "hello world python programming"
char_count = Counter(text)
print("문자 빈도:", char_count)

# 단어 빈도 분석
words = text.split()
word_count = Counter(words)
print("단어 빈도:", word_count)

# 가장 흔한 요소 찾기
most_common_chars = char_count.most_common(3)
print("가장 흔한 문자 3개:", most_common_chars)

# 카운터 연산
counter1 = Counter("aabbcc")
counter2 = Counter("aabbccdd")
print("교집합:", counter1 & counter2)
print("합집합:", counter1 | counter2)
print("차집합:", counter1 - counter2)

실용적인 Counter 활용

# 웹 로그 분석 시뮬레이션
log_data = [
    "192.168.1.1", "192.168.1.2", "192.168.1.1",
    "10.0.0.1", "192.168.1.1", "10.0.0.1",
    "192.168.1.2", "192.168.1.1"
]

ip_counter = Counter(log_data)
print("IP 접속 빈도:")
for ip, count in ip_counter.most_common():
    print(f"  {ip}: {count}회")

# 임계값 이상 접속한 IP 찾기
suspicious_ips = [ip for ip, count in ip_counter.items() if count >= 3]
print(f"의심스러운 IP: {suspicious_ips}")

deque: 양방향 큐의 위력

deque (덱)는 양쪽 끝에서 빠른 추가/삭제가 가능한 리스트형 컨테이너입니다. 스택과 큐를 효율적으로 구현할 수 있습니다.

from collections import deque
import time

# deque vs list 성능 비교
def performance_test():
    # 리스트에서 왼쪽 삽입/삭제 (비효율적)
    normal_list = []
    start_time = time.time()
    for i in range(10000):
        normal_list.insert(0, i)  # O(n) 연산
    list_time = time.time() - start_time
    
    # deque에서 왼쪽 삽입/삭제 (효율적)
    dq = deque()
    start_time = time.time()
    for i in range(10000):
        dq.appendleft(i)  # O(1) 연산
    deque_time = time.time() - start_time
    
    print(f"List 왼쪽 삽입: {list_time:.4f}초")
    print(f"Deque 왼쪽 삽입: {deque_time:.4f}초")
    print(f"성능 차이: {list_time/deque_time:.0f}배")

performance_test()

deque 활용 예시

# 최근 N개 항목 유지하는 링 버퍼
class RecentHistory:
    def __init__(self, maxlen=5):
        self.history = deque(maxlen=maxlen)
    
    def add_item(self, item):
        self.history.append(item)
    
    def get_recent(self):
        return list(self.history)

# 사용 예시
history = RecentHistory(maxlen=3)
for i in range(7):
    history.add_item(f"작업-{i}")
    print(f"현재 히스토리: {history.get_recent()}")

# 회전 기능 활용
dq = deque([1, 2, 3, 4, 5])
print(f"원본: {dq}")
dq.rotate(2)  # 오른쪽으로 2칸 회전
print(f"오른쪽 2칸 회전: {dq}")
dq.rotate(-3)  # 왼쪽으로 3칸 회전
print(f"왼쪽 3칸 회전: {dq}")

namedtuple: 가독성 있는 튜플

namedtuple은 필드명을 가진 튜플을 생성하는 팩토리 함수입니다. 클래스보다 가볍지만 일반 튜플보다 가독성이 뛰어납니다.

from collections import namedtuple

# Point 클래스 정의
Point = namedtuple('Point', ['x', 'y'])

# 사용 예시
p1 = Point(10, 20)
p2 = Point(30, 40)

print(f"점 p1: {p1}")
print(f"p1.x: {p1.x}, p1.y: {p1.y}")
print(f"p1[0]: {p1[0]}, p1[1]: {p1[1]}")  # 인덱스로도 접근 가능

# 거리 계산
def distance(p1, p2):
    return ((p2.x - p1.x)**2 + (p2.y - p1.y)**2) ** 0.5

print(f"두 점 사이 거리: {distance(p1, p2)}")

# _replace 메서드 (불변 객체지만 새로운 인스턴스 생성)
p3 = p1._replace(x=50)
print(f"p1 수정된 버전: {p3}")

실용적인 namedtuple 활용

# 학생 정보 관리
Student = namedtuple('Student', ['name', 'age', 'grade', 'subjects'])

students = [
    Student('김철수', 20, 'A+', ['수학', '물리']),
    Student('이영희', 19, 'A', ['화학', '생물']),
    Student('박민수', 21, 'B+', ['수학', '화학'])
]

# 우수 학생 찾기
excellent_students = [s for s in students if s.grade in ['A+', 'A']]
print("우수 학생:")
for student in excellent_students:
    print(f"  {student.name} ({student.age}세) - {student.grade}")

# 수학을 듣는 학생 찾기
math_students = [s for s in students if '수학' in s.subjects]
print("\n수학 수강생:")
for student in math_students:
    print(f"  {student.name}: {', '.join(student.subjects)}")

defaultdict: 키 오류 없는 딕셔너리

defaultdict는 존재하지 않는 키에 접근할 때 기본값을 자동으로 생성하는 딕셔너리입니다. KeyError를 방지하고 코드를 간결하게 만듭니다.

from collections import defaultdict

# 기본 딕셔너리 vs defaultdict 비교
def normal_dict_example():
    data = {}
    items = ['apple', 'banana', 'apple', 'cherry', 'banana', 'apple']
    
    for item in items:
        if item in data:
            data[item] += 1
        else:
            data[item] = 1
    return data

def defaultdict_example():
    data = defaultdict(int)  # 기본값 0
    items = ['apple', 'banana', 'apple', 'cherry', 'banana', 'apple']
    
    for item in items:
        data[item] += 1  # KeyError 걱정 없음
    return data

print("일반 딕셔너리:", normal_dict_example())
print("defaultdict:", dict(defaultdict_example()))

defaultdict의 다양한 활용

# 그룹화 작업
from collections import defaultdict

# 학생들을 학년별로 그룹화
students = [
    ('김철수', '1학년'), ('이영희', '2학년'), ('박민수', '1학년'),
    ('최지은', '3학년'), ('정우성', '2학년'), ('한가인', '1학년')
]

grade_groups = defaultdict(list)
for name, grade in students:
    grade_groups[grade].append(name)

for grade, students in grade_groups.items():
    print(f"{grade}: {', '.join(students)}")

# 중첩 defaultdict
nested_dd = defaultdict(lambda: defaultdict(int))
transactions = [
    ('2024', '01', 100), ('2024', '01', 200),
    ('2024', '02', 150), ('2023', '12', 300)
]

for year, month, amount in transactions:
    nested_dd[year][month] += amount

print("\n연도별 월별 거래액:")
for year in nested_dd:
    print(f"  {year}년:")
    for month, total in nested_dd[year].items():
        print(f"    {month}월: {total}")

실습: 로그 분석 도구

collections 모듈의 여러 데이터 구조를 조합하여 실용적인 로그 분석 도구를 만들어보겠습니다.

from collections import Counter, defaultdict, deque, namedtuple
from datetime import datetime

# 로그 엔트리 구조 정의
LogEntry = namedtuple('LogEntry', ['timestamp', 'level', 'message', 'ip'])

class LogAnalyzer:
    def __init__(self, max_recent=100):
        self.recent_logs = deque(maxlen=max_recent)  # 최근 로그만 유지
        self.level_counts = Counter()                 # 로그 레벨별 카운트
        self.ip_activity = defaultdict(list)          # IP별 활동 기록
        self.error_patterns = Counter()               # 에러 패턴 분석
    
    def add_log(self, timestamp, level, message, ip="unknown"):
        """로그 엔트리 추가"""
        log_entry = LogEntry(timestamp, level, message, ip)
        
        # 최근 로그에 추가
        self.recent_logs.append(log_entry)
        
        # 레벨별 카운트
        self.level_counts[level] += 1
        
        # IP별 활동 기록
        self.ip_activity[ip].append((timestamp, level, message))
        
        # 에러 패턴 분석
        if level in ['ERROR', 'CRITICAL']:
            # 에러 메시지에서 패턴 추출 (단순화)
            words = message.lower().split()
            for word in words:
                if len(word) > 3:  # 의미있는 단어만
                    self.error_patterns[word] += 1
    
    def get_stats(self):
        """통계 정보 반환"""
        return {
            'total_logs': len(self.recent_logs),
            'level_distribution': dict(self.level_counts),
            'unique_ips': len(self.ip_activity),
            'top_error_patterns': self.error_patterns.most_common(5),
            'most_active_ips': [
                (ip, len(activities)) 
                for ip, activities in sorted(
                    self.ip_activity.items(), 
                    key=lambda x: len(x[1]), 
                    reverse=True
                )[:5]
            ]
        }
    
    def get_recent_errors(self):
        """최근 에러 로그만 반환"""
        return [
            log for log in self.recent_logs 
            if log.level in ['ERROR', 'CRITICAL']
        ]

# 사용 예시
analyzer = LogAnalyzer(max_recent=50)

# 샘플 로그 데이터 추가
sample_logs = [
    ("2024-01-01 10:00:00", "INFO", "User login successful", "192.168.1.100"),
    ("2024-01-01 10:05:00", "WARNING", "High memory usage detected", "192.168.1.100"),
    ("2024-01-01 10:10:00", "ERROR", "Database connection failed", "192.168.1.101"),
    ("2024-01-01 10:15:00", "INFO", "User logout", "192.168.1.100"),
    ("2024-01-01 10:20:00", "CRITICAL", "System disk full", "192.168.1.102"),
    ("2024-01-01 10:25:00", "ERROR", "Database timeout error", "192.168.1.101"),
    ("2024-01-01 10:30:00", "INFO", "User login successful", "192.168.1.103"),
]

for timestamp, level, message, ip in sample_logs:
    analyzer.add_log(timestamp, level, message, ip)

# 분석 결과 출력
print("=== 로그 분석 결과 ===")
stats = analyzer.get_stats()
print(f"총 로그 수: {stats['total_logs']}")
print(f"고유 IP 수: {stats['unique_ips']}")

print("\n레벨별 분포:")
for level, count in stats['level_distribution'].items():
    print(f"  {level}: {count}개")

print("\n활발한 IP Top 5:")
for ip, count in stats['most_active_ips']:
    print(f"  {ip}: {count}개 요청")

print("\n에러 패턴 Top 5:")
for pattern, count in stats['top_error_patterns']:
    print(f"  '{pattern}': {count}회")

print("\n최근 에러 로그:")
for error_log in analyzer.get_recent_errors():
    print(f"  {error_log.timestamp} [{error_log.level}] {error_log.message} (from {error_log.ip})")

코드 실행 결과

=== 로그 분석 결과 ===
총 로그 수: 7
고유 IP 수: 4

레벨별 분포:
  INFO: 3개
  WARNING: 1개
  ERROR: 2개
  CRITICAL: 1개

활발한 IP Top 5:
  192.168.1.100: 3개 요청
  192.168.1.101: 2개 요청
  192.168.1.102: 1개 요청
  192.168.1.103: 1개 요청

에러 패턴 Top 5:
  'database': 2회
  'connection': 1회
  'failed': 1회
  'timeout': 1회
  'error': 1회

마무리

collections 모듈의 고급 데이터 구조들은 각각 특별한 용도에 최적화되어 있습니다. Counter는 카운팅, deque는 양방향 큐, namedtuple은 가독성 있는 데이터 구조, defaultdict는 안전한 딕셔너리 접근을 제공합니다.

적절한 데이터 구조 선택은 코드의 성능과 가독성을 크게 향상시킵니다. 다음 강의에서는 타입 힌트를 통한 코드 안전성 향상 방법을 학습하겠습니다.