데이터 필터링과 집계

완성 목표: 원본을 지우지 않는 작은 분석 파이프라인

이번 실습은 list 안의 dict를 입력으로 받아 검증, 정규화, 필터링, 집계, 정렬 순서로 보고서를 만든다. 유효하지 않은 행을 조용히 버리지 않고 행 번호와 거부 이유를 남기는 것이 핵심이다. 모든 금액은 int로 처리하며 외부 파일이나 패키지가 없어도 전체 코드를 그대로 실행할 수 있다.

1단계: 직원 5명의 조건 필터, 튜플 변환, 급여 정렬, 부서별 평균을 만든다.
2단계: 판매 행의 필수 필드와 정확한 타입을 검사하고 문자열을 정규화한다.
3단계: generator로 고액 매출을 지연 필터링하고 Counter와 defaultdict로 여러 관점의 집계를 만든다.

1단계: 직원 데이터로 기본 연산 확인하기

컴프리헨션은 기존 iterable에서 조건을 만족하는 원소를 고르거나 새 표현으로 바꿔 list를 만드는 데 알맞다. 복잡한 검증처럼 여러 분기와 부수 효과가 필요하면 일반 for 문이 더 읽기 쉽다. Python 공식 튜토리얼의 list comprehension 설명도 같은 구성 순서를 제시한다 (2026-08-03 확인).

from collections import defaultdict

employees = [
    {"name": "Alice", "department": "Engineering", "age": 30, "salary": 85000},
    {"name": "Bob", "department": "Marketing", "age": 25, "salary": 60000},
    {"name": "Charlie", "department": "Engineering", "age": 35, "salary": 95000},
    {"name": "David", "department": "HR", "age": 45, "salary": 70000},
    {"name": "Eve", "department": "Engineering", "age": 28, "salary": 78000},
]

engineering_high_paid = [
    employee["name"]
    for employee in employees
    if employee["department"] == "Engineering" and employee["salary"] >= 80000
]
age_30_or_more = [
    (employee["name"], employee["department"])
    for employee in employees
    if employee["age"] >= 30
]
top_three = sorted(employees, key=lambda employee: employee["salary"], reverse=True)[:3]

salaries_by_department = defaultdict(list)
for employee in employees:
    salaries_by_department[employee["department"]].append(employee["salary"])
average_salary = {
    department: sum(salaries) // len(salaries)
    for department, salaries in sorted(salaries_by_department.items())
}

print("Engineering 80000+:", engineering_high_paid)
print("Age 30+:", age_30_or_more)
print("Top 3:", [(employee["name"], employee["salary"]) for employee in top_three])
print("Average:", average_salary)

Engineering 80000+: ['Alice', 'Charlie']
Age 30+: [('Alice', 'Engineering'), ('Charlie', 'Engineering'), ('David', 'HR')]
Top 3: [('Charlie', 95000), ('Alice', 85000), ('Eve', 78000)]
Average: {'Engineering': 86000, 'HR': 70000, 'Marketing': 60000}

sorted는 원본 employees를 바꾸지 않고 새 list를 반환한다. key에는 비교 기준을, reverse=True에는 내림차순을 지정했다. Python 내장 함수 sorted 문서에 따르면 정렬은 안정적이므로 key가 같은 항목의 기존 순서가 유지된다 (2026-08-03 확인). 평균은 이 예제에서 모두 정수로 나누어떨어지므로 // 결과가 정확하지만, 실제 평균에 소수 부분이 필요하다면 / 또는 Decimal 등 요구 정밀도에 맞는 표현을 정해야 한다.

2단계: 판매 원본과 검증 규칙 정하기

분석 전에 입력 계약부터 정한다. id·region·category·month·amount가 모두 있어야 하며 문자열은 빈칸이 아니어야 한다. bool은 int의 하위 타입이므로 type(amount) is int로 검사해 True가 금액 1로 통과하는 일을 막는다. amount는 0보다 커야 하고, 유효한 문자열은 양끝 공백 제거 후 표기 형식을 통일한다.

REQUIRED_FIELDS = ("id", "region", "category", "month", "amount")

def validate_and_normalize(row, row_number):
    if not isinstance(row, dict):
        return None, f"row {row_number}: dict가 아님"
    missing = [field for field in REQUIRED_FIELDS if field not in row]
    if missing:
        return None, f"row {row_number}: 필수 필드 누락 {missing}"
    for field in ("id", "region", "category", "month"):
        if not isinstance(row[field], str) or not row[field].strip():
            return None, f"row {row_number}: {field}는 비어 있지 않은 str이어야 함"
    if type(row["amount"]) is not int or row["amount"] <= 0:
        return None, f"row {row_number}: amount는 0보다 큰 int여야 함"
    normalized = {
        "id": row["id"].strip(),
        "region": row["region"].strip().title(),
        "category": row["category"].strip().title(),
        "month": row["month"].strip(),
        "amount": row["amount"],
    }
    return normalized, None

원본 dict를 직접 수정하지 않고 normalized를 새로 만든다. 이렇게 하면 정제 전후 값을 비교할 수 있고 재실행 결과도 예측하기 쉽다. 실무에서는 month가 YYYY-MM인지, id가 중복인지 같은 도메인 규칙도 추가한다.

3단계: 유효 행과 거부 행을 함께 보존하기

def partition_rows(rows):
    valid_rows = []
    rejected_rows = []
    for row_number, row in enumerate(rows, start=1):
        normalized, error = validate_and_normalize(row, row_number)
        if error is None:
            valid_rows.append(normalized)
        else:
            rejected_rows.append({"row_number": row_number, "reason": error, "raw": row})
    return valid_rows, rejected_rows

잘못된 행도 raw와 reason을 함께 남긴다. 다만 로그나 화면에 원본을 노출할 때는 개인정보와 비밀값을 마스킹해야 한다. 이 예제의 입력에는 그런 값이 없다. 입력량이 매우 크다면 거부 행 전체 대신 별도 저장소나 오류 건수·샘플만 보관하는 정책도 필요하다.

4단계: generator와 집계 자료구조 배치하기

amount가 1000을 초과한 행은 yield로 하나씩 내보낸다. 호출 즉시 전체 결과를 만들지 않고 소비할 때 진행되는 generator의 동작은 Python 공식 튜토리얼의 Generators에서 확인할 수 있다 (2026-08-03 확인). 이 iterator는 한 번 소비하면 끝나므로 보고서 항목마다 high_value_sales(valid_rows)를 새로 호출한다.

from collections import Counter, defaultdict

def high_value_sales(rows):
    for row in rows:
        if row["amount"] > 1000:
            yield row

def aggregate_sales(valid_rows):
    sales_at_least_1000 = [row for row in valid_rows if row["amount"] >= 1000]
    totals_by_region = defaultdict(int)
    amounts_by_category = defaultdict(list)
    totals_by_month_category = defaultdict(int)
    for row in valid_rows:
        totals_by_region[row["region"]] += row["amount"]
        amounts_by_category[row["category"]].append(row["amount"])
        totals_by_month_category[(row["month"], row["category"])] += row["amount"]
    counts_by_region = Counter(row["region"] for row in valid_rows)
    top_three = sorted(valid_rows, key=lambda row: (-row["amount"], row["id"]))[:3]
    return {
        "sales_at_least_1000": sales_at_least_1000,
        "totals_by_region": dict(sorted(totals_by_region.items())),
        "counts_by_region": dict(sorted(counts_by_region.items())),
        "amounts_by_category": dict(sorted(amounts_by_category.items())),
        "totals_by_month_category": dict(sorted(totals_by_month_category.items())),
        "high_value_ids": [row["id"] for row in high_value_sales(valid_rows)],
        "top_three": [(row["id"], row["amount"]) for row in top_three],
    }

Counter는 hashable 값의 빈도를 세는 dict 하위 클래스이고, defaultdict는 없는 key를 처음 조회할 때 factory로 기본값을 만든다. Counter 공식 문서와 defaultdict 공식 문서에서 각각의 동작을 확인할 수 있다 (2026-08-03 확인). 단순 빈도에는 Counter, key별 합계나 list 그룹에는 defaultdict가 의도를 잘 드러낸다.

메모리 숫자를 과장하지 않기

from sys import getsizeof

eager = [number * 2 for number in range(10_000)]
lazy = (number * 2 for number in range(10_000))
print("list container bytes:", getsizeof(eager))
print("generator object bytes:", getsizeof(lazy))

출력 바이트 수는 Python 버전과 구현에 따라 달라지므로 고정 예상값을 적지 않는다. 더 중요한 한계는 getsizeof가 객체에 직접 귀속된 크기만 세고 그 객체가 참조하는 원소들의 전체 크기는 포함하지 않는다는 점이다. 따라서 위 코드는 list 컨테이너 본체와 generator 객체의 얕은 크기 비교일 뿐, 파이프라인 전체 메모리 측정이 아니다. 이 제한은 sys.getsizeof 공식 문서에도 명시되어 있다 (2026-08-03 확인). 실제 판단에는 대표 입력을 사용한 peak memory 측정이 필요하다.

마지막 완성 프로그램

아래 코드는 검증 가능한 유효 행 6개와 의도적으로 잘못 만든 행 3개를 함께 처리한다. 출력 순서를 재현할 수 있도록 보고용 dict와 복합 key를 정렬했다.

from collections import Counter, defaultdict

REQUIRED_FIELDS = ("id", "region", "category", "month", "amount")

SALES = [
    {"id": "S-001", "region": " north ", "category": "hardware", "month": "2026-01", "amount": 1200},
    {"id": "S-002", "region": "south", "category": "software", "month": "2026-01", "amount": 900},
    {"id": "S-003", "region": "north", "category": "software", "month": "2026-02", "amount": 1500},
    {"id": "S-004", "region": "west", "category": "hardware", "month": "2026-02", "amount": 700},
    {"id": "S-005", "region": "south", "category": "service", "month": "2026-02", "amount": 2000},
    {"id": "S-006", "region": "west", "category": "software", "month": "2026-01", "amount": 1000},
    {"id": "S-007", "region": "north", "category": "service", "month": "2026-03"},
    {"id": "S-008", "region": "south", "category": "hardware", "month": "2026-03", "amount": "1300"},
    {"id": "S-009", "region": " ", "category": "service", "month": "2026-03", "amount": 500},
]

def validate_and_normalize(row, row_number):
    if not isinstance(row, dict):
        return None, f"row {row_number}: dict가 아님"
    missing = [field for field in REQUIRED_FIELDS if field not in row]
    if missing:
        return None, f"row {row_number}: 필수 필드 누락 {missing}"
    for field in ("id", "region", "category", "month"):
        if not isinstance(row[field], str) or not row[field].strip():
            return None, f"row {row_number}: {field}는 비어 있지 않은 str이어야 함"
    if type(row["amount"]) is not int or row["amount"] <= 0:
        return None, f"row {row_number}: amount는 0보다 큰 int여야 함"
    return {
        "id": row["id"].strip(),
        "region": row["region"].strip().title(),
        "category": row["category"].strip().title(),
        "month": row["month"].strip(),
        "amount": row["amount"],
    }, None

def partition_rows(rows):
    valid_rows = []
    rejected_rows = []
    for row_number, row in enumerate(rows, start=1):
        normalized, error = validate_and_normalize(row, row_number)
        if error is None:
            valid_rows.append(normalized)
        else:
            rejected_rows.append({"row_number": row_number, "reason": error, "raw": row})
    return valid_rows, rejected_rows

def high_value_sales(rows):
    for row in rows:
        if row["amount"] > 1000:
            yield row

def aggregate_sales(valid_rows):
    sales_at_least_1000 = [row for row in valid_rows if row["amount"] >= 1000]
    totals_by_region = defaultdict(int)
    amounts_by_category = defaultdict(list)
    totals_by_month_category = defaultdict(int)
    for row in valid_rows:
        totals_by_region[row["region"]] += row["amount"]
        amounts_by_category[row["category"]].append(row["amount"])
        totals_by_month_category[(row["month"], row["category"])] += row["amount"]
    counts_by_region = Counter(row["region"] for row in valid_rows)
    top_three = sorted(valid_rows, key=lambda row: (-row["amount"], row["id"]))[:3]
    return {
        "sales_at_least_1000": sales_at_least_1000,
        "totals_by_region": dict(sorted(totals_by_region.items())),
        "counts_by_region": dict(sorted(counts_by_region.items())),
        "amounts_by_category": dict(sorted(amounts_by_category.items())),
        "totals_by_month_category": dict(sorted(totals_by_month_category.items())),
        "high_value_ids": [row["id"] for row in high_value_sales(valid_rows)],
        "top_three": [(row["id"], row["amount"]) for row in top_three],
    }

def print_report(valid_rows, rejected_rows, report):
    print(f"Valid: {len(valid_rows)}, Rejected: {len(rejected_rows)}")
    for rejected in rejected_rows:
        print("REJECT", rejected["reason"])
    print("Amount >= 1000:", [row["id"] for row in report["sales_at_least_1000"]])
    print("Region totals:", report["totals_by_region"])
    print("Region counts:", report["counts_by_region"])
    print("Category amounts:", report["amounts_by_category"])
    print("Month/category totals:")
    for (month, category), total in report["totals_by_month_category"].items():
        print(f"  {month} | {category}: {total}")
    print("Amount > 1000 generator:", report["high_value_ids"])
    print("Top 3:", report["top_three"])

valid_rows, rejected_rows = partition_rows(SALES)
report = aggregate_sales(valid_rows)
print_report(valid_rows, rejected_rows, report)

Valid: 6, Rejected: 3
REJECT row 7: 필수 필드 누락 ['amount']
REJECT row 8: amount는 0보다 큰 int여야 함
REJECT row 9: region는 비어 있지 않은 str이어야 함
Amount >= 1000: ['S-001', 'S-003', 'S-005', 'S-006']
Region totals: {'North': 2700, 'South': 2900, 'West': 1700}
Region counts: {'North': 2, 'South': 2, 'West': 2}
Category amounts: {'Hardware': [1200, 700], 'Service': [2000], 'Software': [900, 1500, 1000]}
Month/category totals:
  2026-01 | Hardware: 1200
  2026-01 | Software: 1900
  2026-02 | Hardware: 700
  2026-02 | Service: 2000
  2026-02 | Software: 1500
Amount > 1000 generator: ['S-001', 'S-003', 'S-005']
Top 3: [('S-005', 2000), ('S-003', 1500), ('S-001', 1200)]

결과를 읽는 순서

먼저 Valid와 Rejected 합이 입력 9건과 같은지 확인한다. 이 수가 다르면 처리 과정에서 행이 사라졌다는 뜻이다.
경계값을 확인한다. S-006의 amount는 정확히 1000이므로 >= 1000 목록에는 들지만 > 1000 generator 목록에는 들지 않는다.
지역 총합의 합 7300과 카테고리 금액 전체의 합 7300이 같은지 교차 검증한다. 서로 다른 집계 축의 총합이 다르면 누락이나 중복을 의심한다.
top 3가 2000, 1500, 1200 순인지 확인한다. 동액일 때는 id 오름차순을 두 번째 기준으로 사용해 출력이 결정적이다.

학습 체크포인트

☐ 필터 조건의 >=와 > 경계값 차이를 테스트했다.

☐ 원본을 수정하지 않고 정규화한 새 dict를 만들었다.

☐ 유효 행 수와 거부 행 수, 각 거부 이유를 확인했다.

☐ 빈도에는 Counter, 합계와 그룹에는 알맞은 defaultdict factory를 선택했다.

☐ generator가 일회성임을 알고 필요할 때 새로 생성했다.

☐ getsizeof 결과를 전체 메모리 사용량으로 해석하지 않았다.

확장 과제

id 중복을 set으로 검출하고 두 번째 행부터 거부한다. 어느 행과 충돌했는지도 이유에 남긴다.
month를 datetime.strptime으로 검증해 2026-13 같은 값을 거부하고 월별 총합을 시간순으로 출력한다.
정답 확인용 assert를 추가한다. 유효 6건, 총액 7300, South 총액 2900, top 1이 S-005인지 검사한다.
대량 입력을 가정해 유효 행도 list로 모으지 않는 스트리밍 버전을 설계한다. 단, 여러 집계를 한 번의 순회에서 모두 갱신하고 거부 기록의 보존 한도를 정한다.

다음 글: 함수로 분석 단계를 분리하는 법

이번 실습에서는 검증, 분할, 집계, 출력 책임을 함수로 나누었다. 다음 글에서는 parameter와 return, scope, 가변 기본 인자 같은 함수 설계 원칙을 정리해 재사용하고 테스트하기 쉬운 분석 코드를 만든다.

참고 자료

Python 공식 튜토리얼: List Comprehensions — 조건 필터와 변환을 포함한 list 생성 문법 (2026-08-03 확인)
Python 내장 함수: sorted — 새 정렬 list, key·reverse와 안정 정렬 (2026-08-03 확인)
Python 표준 라이브러리: Counter — hashable 객체의 빈도 계산 (2026-08-03 확인)
Python 표준 라이브러리: defaultdict — factory로 누락 key의 기본값을 만드는 dict 하위 클래스 (2026-08-03 확인)
Python 공식 튜토리얼: Generators — yield와 실행 상태 보존 (2026-08-03 확인)
Python 표준 라이브러리: sys.getsizeof — 얕은 객체 크기와 참조 대상 제외 범위 (2026-08-03 확인)

댓글을 불러오는 중…