대규모 데이터셋을 처리할 때, 데이터를 읽고, 레코드를 필터링하고, 값을 변환하고, 결과를 다른 곳에 쓰는 것과 같은 여러 작업을 체인(chain) 형태로 연결하는 것은 매우 일반적입니다.

직관적인 구현 방식은 각 단계 이후에 중간 리스트(intermediate lists)를 생성하는 경우가 많습니다:

data = read_data()
filtered = filter_data(data)
transformed = transform_data(filtered)
...

이 방식은 작은 데이터셋에는 잘 작동하지만, 모든 중간 결과가 메모리에 저장되기 때문에 수백만 개의 레코드를 처리할 때는 비효율적이 됩니다.

더 나은 접근 방식은 **제너레이터 기반 파이프라인(generator-based pipelines)**을 사용하는 것입니다.

제너레이터(Generators)를 사용하면 데이터가 각 처리 단계마다 한 번에 하나씩 흐를 수 있어, 코드를 더 메모리 효율적(memory-efficient)이고, 조합 가능(composable)하며, 유지보수하기 쉽게 만들어 줍니다.

이 글에서는 간단한 텍스트 처리부터 ETL 및 컴퓨터 비전(computer vision) 워크로드에 이르기까지 여러 가지 제너레이터 파이프라인을 구축해 보겠습니다.

제너레이터 파이프라인이란 무엇인가?

제너레이터 파이프라인은 여러 개의 독립적인 단계(stages)로 구성됩니다.

각 단계는 다음과 같이 동작합니다:

• 반복 가능한 객체(iterable, 보통 다른 제너레이터)를 전달받습니다.
• 각 항목(item)을 처리합니다.
• yield를 사용하여 항목을 생성합니다.

한 단계의 출력이 다음 단계의 입력이 됩니다.

가장 큰 장점 중 하나는 일반적으로 한 번에 단 하나의 항목만이 파이프라인을 통해 이동한다는 점입니다.

예제 1: 대규모 텍스트 파일 처리

수 기가바이트(multi-gigabyte) 크기의 로그 파일이 있고 다음과 같은 작업을 수행하고 싶다고 가정해 봅시다:

• 주석 무시
• 공백 제거
• 모든 내용을 대문자로 변환

각 단계는 그 자체로 하나의 제너레이터가 됩니다.

def read_lines(filename):
    with open(filename) as f:
        for line in f:
...

이들을 연결하는 방법은 간단합니다:

pipeline = uppercase(
    strip_lines(
        remove_comments(
...

최종 루프가 제너레이터를 소비(consuming)하기 시작할 때까지는 아무 일도 일어나지 않는다는 점에 주목하세요.

어느 시점에서든, 단 한 줄만이 처리되고 있습니다.

예제 2: 간단한 ETL 파이프라인 구축

제너레이터 파이프라인 (Generator pipelines)은 ETL (Extract, Transform, Load; 추출, 변환, 적재)에 매우 적합합니다.

간단한 모델을 정의해 보겠습니다.

from dataclasses import dataclass

@dataclass
...

추출 (Extract)

def extract():
    users = [
        {"id": 1, "name": "Alice", "age": 22},
...

필터링 (Filter)

def filter_adults(users):
    for user in users:
        if user["age"] >= 18:
...

변환 (Transform)

def transform(users):
    for user in users:
        yield User(**user)

적재 (Load)

def load(users):
    for user in users:
        print(user)

모든 단계 결합하기

load(
    transform(
        filter_adults(
...

출력 결과:

User(id=1, name='Alice', age=22)
User(id=3, name='Charlie', age=35)

각 단계는 단일 책임 (single responsibility)을 가지므로, 파이프라인을 확장하고 테스트하기가 용이합니다.

예제 3: 컴퓨터 비전 (Computer Vision) 파이프라인

제너레이터 파이프라인은 이미지가 상당한 메모리를 소비할 수 있는 머신러닝 (machine learning) 및 컴퓨터 비전 (computer vision) 분야에서 특히 유용합니다.

def load_images(paths):
    for path in paths:
        image = cv2.imread(path)
...

사용법:

predictions = predict(
    normalize(
        resize(
...

모든 이미지를 먼저 메모리에 로드하는 대신, 이미지는 한 번에 하나씩 로드되고, 처리되며, 예측됩니다.

예제 4: 데이터베이스 레코드 처리

동일한 접근 방식이 데이터베이스 쿼리 (database queries)에도 적용됩니다.

def fetch_rows(cursor):
    while True:
        rows = cursor.fetchmany(1000)
...

레코드 정제 (Cleaning):

def clean(rows):
    for row in rows:
        row["name"] = row["name"].strip()
...

저장하기:

def save(rows):
    for row in rows:
        insert_into_new_db(row)

데이터베이스에서는 작은 배치 (batch) 단위로만 가져오며, 개별 행 (row)들은 파이프라인을 통해 계속해서 흘러갑니다.

yield from 사용하기

때때로 제너레이터는 단순히 다른 반복 가능한 객체 (iterable)를 전달하기만 할 때가 있습니다.

수동으로 반복하는 대신:

def flatten(data):
    for lst in data:
        for item in lst:
...

yield from을 사용하여 위임을 수행할 수 있습니다.

def flatten(data):
    for lst in data:
        yield from lst

예시:

data = [[1, 2], [3, 4], [5]]

print(list(flatten(data)))

출력:

[1, 2, 3, 4, 5]

yield from은 제너레이터 합성 (generator composition)을 더 깔끔하고 읽기 쉽게 만들어 줍니다.

제너레이터 표현식 (Generator Expressions)

단순한 변환의 경우, 제너레이터 표현식 (generator expressions)은 간결하고 표현력이 좋습니다.

numbers = range(10)

pipeline = (
...

출력:

[0, 4, 16, 36, 64]

이 방식은 가벼운 파이프라인 (lightweight pipelines)에 특히 효과적입니다.

구성 가능한 파이프라인 구축하기

애플리케이션이 성장함에 따라, 단계 (stages)를 동적으로 합성하는 것이 유용합니다.

from collections.abc import Iterable, Callable

def run_pipeline(data: Iterable, *stages: Callable):
...

이제 런타임 (runtime)에 파이프라인을 정의합니다.

pipeline = run_pipeline(
    range(20),
    lambda xs: (x for x in xs if x % 2 == 0),
...

출력:

[1, 5, 17, 37, 65, 101, 145, 197, 257, 325]

이를 통해 기존 코드를 수정하지 않고도 처리 단계 (processing stages)를 교체, 재정렬 또는 추가하기가 쉬워집니다.

왜 제너레이터 파이프라인을 사용하는가?

제너레이터 파이프라인은 리스트 기반 처리 (list-based processing)에 비해 여러 가지 장점을 제공합니다.

• 메모리 효율성 (Memory efficient) — 중간 컬렉션 (intermediate collections)을 저장하는 대신 한 번에 하나의 항목만 처리합니다.
• 지연 평가 (Lazy evaluation) — 데이터가 요청될 때만 작업이 수행됩니다.
• 합성 가능성 (Composable) — 각 단계는 단일 책임 (single responsibility)을 수행합니다.
• 재사용성 (Reusable) — 단계를 파이프라인 전반에 걸쳐 혼합하거나 조합할 수 있습니다.
• 확장성 (Scalable) — 파일, 데이터베이스 커서 (database cursors), 스트림 (streams), API, 그리고 머신러닝 데이터셋에 이상적입니다.
• 테스트 가능성 (Testable) — 각 제너레이터는 독립적으로 단위 테스트 (unit test)를 수행할 수 있습니다.

이 패턴이 사용되는 곳

제너레이터 파이프라인은 다음과 같은 많은 실제 시스템에서 나타납니다:

• ETL 및 데이터 엔지니어링 워크플로우 (Workflows)
• 로그 처리 (Log processing)
• 웹 스크래핑 (Web scraping)
• CSV 및 JSON 처리
• 컴퓨터 비전 전처리 (Computer vision preprocessing)
• 머신러닝 데이터 로더 (Machine learning data loaders)
• 스트리밍 애플리케이션 (Streaming applications)
• API 응답 처리 (API response processing)

이 패턴을 인식하기 시작하면, 수많은 Python 프로젝트에서 이를 활용할 기회를 발견하게 될 것입니다.

결론

Python 제너레이터 (Generators)는 단순히 데이터를 반복 (Iterate) 하는 수단 그 이상입니다. 이는 깨끗하고, 모듈화되어 있으며, 메모리 효율적인 처리 파이프라인 (Pipelines)을 구축하기 위한 강력한 도구입니다.

워크플로우를 작은 제너레이터 단계로 분리함으로써, 코드를 이해하기 쉽고, 테스트 가능하며, 확장 가능하게 유지하면서도 사실상 어떤 크기의 데이터셋도 처리할 수 있습니다.

대용량 파일, 데이터베이스, API 또는 AI 워크로드 (Workloads)를 다루고 있다면, 제너레이터 파이프라인은 여러분의 Python 도구 상자에 반드시 추가해야 할 기술입니다.

즐거운 코딩 되세요! 🚀