Skip to content

Python 제너레이터: yield, 제너레이터 표현식, 지연 평가 완전 가이드

Soren Atelier
Name
Soren Atelier

Updated on

10GB 로그 파일을 처리하거나 수백만 개의 데이터베이스 레코드를 스트리밍하는 작업은 Python 애플리케이션을 쉽게 마비시킬 수 있습니다. 전통적인 방식처럼 모든 데이터를 한 번에 메모리에 올리면 성능 병목, 메모리 에러, 불만족스러운 사용자 경험으로 이어지기 쉽습니다. 이때 Python 제너레이터가 핵심이 됩니다. 제너레이터는 모든 값을 미리 저장하는 대신 필요할 때(on-demand) 값들을 생성하므로, 메모리 사용량을 최소화한 채로 초대형 데이터셋을 처리할 수 있게 해줍니다.

📚

    Python 제너레이터란 무엇이며, 왜 중요한가

    제너레이터는 모든 값을 한 번에 계산해서 반환하는 것이 아니라, 시간에 따라 값을 하나씩 생성하는 특수한 함수입니다. return으로 단일 결과를 돌려주는 일반 함수와 달리, 제너레이터는 yield 키워드로 값들을 순차적으로 만들어 내며, 각 값을 만든 뒤 실행을 잠시 멈췄다가 다음 값이 필요해지면 다시 이어서 실행합니다.

    제너레이터의 가장 근본적인 장점은 지연 평가(lazy evaluation) 입니다. 즉, 값은 필요할 때만 생성됩니다. 이로 인해 다음 두 가지 큰 이점이 생깁니다.

    1. 메모리 효율성: 제너레이터는 전체 시퀀스를 메모리에 저장하지 않습니다. 10개 숫자를 생성하는 제너레이터나 10억 개 숫자를 생성하는 제너레이터나 메모리 사용량은 거의 동일합니다.
    2. 성능: 전체 데이터셋이 준비될 때까지 기다릴 필요 없이, 첫 번째 yield 값부터 즉시 처리를 시작할 수 있습니다.

    아래는 차이를 보여주는 간단한 비교입니다:

    # Traditional approach - loads entire list into memory
def get_squares_list(n):
    result = []
    for i in range(n):
        result.append(i * i)
    return result
 
# Generator approach - produces values one at a time
def get_squares_generator(n):
    for i in range(n):
        yield i * i
 
# Memory impact comparison
import sys
 
# List approach
squares_list = get_squares_list(1000000)
print(f"List memory: {sys.getsizeof(squares_list):,} bytes")  # ~8,000,000 bytes
 
# Generator approach
squares_gen = get_squares_generator(1000000)
print(f"Generator memory: {sys.getsizeof(squares_gen):,} bytes")  # ~112 bytes

    메모리 차이는 매우 큽니다. 이 예시에서 제너레이터는 리스트보다 99.999% 적은 메모리를 사용합니다. 데이터셋이 커질수록 이 격차는 더 극적으로 커집니다.

    yield 키워드: 제너레이터 함수의 핵심

    yield 키워드는 일반 함수를 제너레이터 함수로 바꿔줍니다. Python이 yield를 만나면, 함수를 즉시 실행해 결과를 만드는 대신 제너레이터 객체를 반환해야 한다는 것을 압니다.

    def countdown(n):
    print(f"Starting countdown from {n}")
    while n > 0:
        yield n
        n -= 1
    print("Countdown complete!")
 
# Creating the generator doesn't execute the function
gen = countdown(3)
print(type(gen))  # <class 'generator'>
 
# Values are produced on-demand
print(next(gen))  # Starting countdown from 3 -> 3
print(next(gen))  # 2
print(next(gen))  # 1
# next(gen)  # Countdown complete! -> Raises StopIteration

    이때 이해해야 할 핵심 동작은 다음과 같습니다.

    • 각 yield에서 실행이 일시 중지되고, 다음 호출 때 정확히 그 지점에서 이어서 실행됩니다.
    • 로컬 변수 상태는 yield 호출 사이에 유지됩니다.
    • 제너레이터 함수가 return(혹은 더 이상 yield할 값이 없음)하면 StopIteration 예외가 발생합니다.

    하나의 제너레이터에 yield는 여러 번 등장할 수 있습니다:

    def data_pipeline():
    # Phase 1: Loading
    yield "Loading data..."
 
    # Phase 2: Processing
    yield "Processing records..."
 
    # Phase 3: Validation
    yield "Validating results..."
 
    # Phase 4: Complete
    yield "Pipeline complete!"
 
for status in data_pipeline():
    print(status)

    제너레이터 프로토콜: iter()와 next() 이해하기

    제너레이터는 두 개의 특수 메서드를 통해 이터레이터 프로토콜을 구현합니다.

    • __iter__(): 이터레이터 객체(즉, 제너레이터 자신)를 반환
    • __next__(): 제너레이터의 다음 값을 반환

    이 덕분에 제너레이터는 for 루프 및 다양한 반복 컨텍스트에서 매우 자연스럽게 동작합니다. 이 프로토콜을 이해하면 내부 동작이 더 명확해집니다.

    def simple_gen():
    yield 1
    yield 2
    yield 3
 
gen = simple_gen()
 
# These are equivalent
print(gen.__next__())  # 1
print(next(gen))       # 2
 
# for loops call __next__() automatically until StopIteration
for value in simple_gen():
    print(value)  # 1, 2, 3

    또한 이터레이터 프로토콜을 직접 구현해 제너레이터와 유사한 동작을 만들 수도 있습니다:

    class CountDown:
    def __init__(self, start):
        self.current = start
 
    def __iter__(self):
        return self
 
    def __next__(self):
        if self.current <= 0:
            raise StopIteration
        self.current -= 1
        return self.current + 1
 
# Behaves like a generator
for num in CountDown(3):
    print(num)  # 3, 2, 1

    하지만 일반적으로 제너레이터 함수가 수동 구현 이터레이터 클래스보다 훨씬 간결하고 읽기 쉽습니다.

    제너레이터 표현식 vs 리스트 컴프리헨션

    제너레이터 표현식은 리스트 컴프리헨션과 비슷하지만, 대괄호 [] 대신 소괄호 ()를 사용해 제너레이터를 만드는 간결한 문법입니다.

    # List comprehension - creates entire list in memory
squares_list = [x * x for x in range(10)]
print(type(squares_list))  # <class 'list'>
print(squares_list)  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
 
# Generator expression - creates generator object
squares_gen = (x * x for x in range(10))
print(type(squares_gen))  # <class 'generator'>
print(squares_gen)  # <generator object at 0x...>
 
# Consume the generator
print(list(squares_gen))  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

    문법 비교:

    FeatureList ComprehensionGenerator Expression
    Syntax[expr for item in iterable](expr for item in iterable)
    ReturnsList objectGenerator object
    MemoryStores all valuesGenerates on-demand
    SpeedFaster for small datasetsFaster for large datasets
    ReusableYes (can iterate multiple times)No (exhausted after one iteration)

    메모리 차이를 보여주는 실용 예시:

    import sys
 
# List comprehension for 1 million numbers
list_comp = [x for x in range(1000000)]
print(f"List comprehension: {sys.getsizeof(list_comp):,} bytes")
 
# Generator expression for the same range
gen_exp = (x for x in range(1000000))
print(f"Generator expression: {sys.getsizeof(gen_exp):,} bytes")
 
# Output:
# List comprehension: 8,000,056 bytes
# Generator expression: 112 bytes

    제너레이터 표현식은 한 번만 순회하면 되고 메모리 사용을 최소화하고 싶을 때 특히 적합합니다.

    yield from: 서브 제너레이터로 위임하기

    yield from 문은 서브 제너레이터나 다른 iterable로의 위임을 단순화합니다. 값을 하나씩 루프를 돌며 yield하는 대신, yield from이 이를 자동으로 처리합니다.

    # Without yield from
def get_numbers_manual():
    for i in range(3):
        yield i
    for i in range(10, 13):
        yield i
 
# With yield from
def get_numbers_delegated():
    yield from range(3)
    yield from range(10, 13)
 
print(list(get_numbers_manual()))      # [0, 1, 2, 10, 11, 12]
print(list(get_numbers_delegated()))   # [0, 1, 2, 10, 11, 12]

    이는 중첩 구조를 평탄화(flatten)할 때 특히 유용합니다:

    def flatten(nested_list):
    for item in nested_list:
        if isinstance(item, list):
            yield from flatten(item)  # Recursive delegation
        else:
            yield item
 
nested = [1, [2, 3, [4, 5]], 6, [7, [8, 9]]]
print(list(flatten(nested)))  # [1, 2, 3, 4, 5, 6, 7, 8, 9]

    또한 yield from은 서브 제너레이터의 예외와 반환값도 올바르게 처리하므로, 복잡한 제너레이터 파이프라인에서 중요한 도구입니다.

    고급: send()와 throw() 메서드

    제너레이터는 단순히 값을 생산하는 것 이상을 할 수 있습니다. send()와 throw()를 통해 값을 “받거나”, 예외를 처리하게 할 수 있어 코루틴 스타일의 양방향 통신도 가능합니다.

    send()로 제너레이터에 값 보내기

    def running_average():
    total = 0
    count = 0
    average = None
 
    while True:
        value = yield average  # Yield current average, receive new value
        total += value
        count += 1
        average = total / count
 
# Create generator
avg = running_average()
next(avg)  # Prime the generator (advance to first yield)
 
# Send values and receive running averages
print(avg.send(10))   # 10.0
print(avg.send(20))   # 15.0
print(avg.send(30))   # 20.0
print(avg.send(40))   # 25.0

    send()는 (1) 제너레이터 내부로 값을 전달하고(이 값이 yield 표현식의 결과가 됨), (2) 다음 yield 지점까지 실행을 진행하는 역할을 동시에 수행합니다.

    throw()로 예외 주입하기

    def error_handling_gen():
    try:
        while True:
            value = yield
            print(f"Received: {value}")
    except ValueError as e:
        print(f"Caught ValueError: {e}")
        yield "Recovered from error"
    except GeneratorExit:
        print("Generator is closing")
 
gen = error_handling_gen()
next(gen)  # Prime the generator
 
gen.send(10)              # Received: 10
gen.send(20)              # Received: 20
result = gen.throw(ValueError, "Invalid value")  # Caught ValueError: Invalid value
print(result)             # Recovered from error
gen.close()               # Generator is closing

    이런 고급 기능들은 상태 머신, 코루틴, 복잡한 비동기 패턴을 구현할 때 특히 유용합니다.

    무한 제너레이터: 끝없는 시퀀스

    제너레이터는 전체 시퀀스를 메모리에 실체화할 필요가 없기 때문에, 무한 시퀀스 생성에 매우 적합합니다.

    # Infinite counter
def count_from(start=0, step=1):
    current = start
    while True:
        yield current
        current += step
 
# Fibonacci sequence
def fibonacci():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b
 
# Cycling through a sequence
def cycle(iterable):
    saved = []
    for item in iterable:
        yield item
        saved.append(item)
    while saved:
        for item in saved:
            yield item
 
# Usage examples
counter = count_from(10, 2)
for _ in range(5):
    print(next(counter))  # 10, 12, 14, 16, 18
 
fib = fibonacci()
print([next(fib) for _ in range(10)])  # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]
 
colors = cycle(['red', 'green', 'blue'])
print([next(colors) for _ in range(8)])  # ['red', 'green', 'blue', 'red', 'green', 'blue', 'red', 'green']

    무한 제너레이터는 이벤트 스트림, 지속적 모니터링, 상태를 갖는 반복 패턴에서 특히 유용합니다.

    제너레이터 체이닝: 데이터 처리 파이프라인 만들기

    제너레이터의 가장 강력한 패턴 중 하나는 여러 제너레이터를 연결해 효율적인 데이터 처리 파이프라인을 구축하는 것입니다. 각 단계는 지연 방식으로 데이터를 처리하고, 중간 결과를 저장하지 않은 채 다음 단계로 전달합니다.

    # Stage 1: Read lines from a file (generator)
def read_log_file(filename):
    with open(filename, 'r') as f:
        for line in f:
            yield line.strip()
 
# Stage 2: Filter lines containing 'ERROR'
def filter_errors(lines):
    for line in lines:
        if 'ERROR' in line:
            yield line
 
# Stage 3: Extract timestamp and message
def parse_error_lines(lines):
    for line in lines:
        parts = line.split(' - ')
        if len(parts) >= 2:
            yield {'timestamp': parts[0], 'message': parts[1]}
 
# Stage 4: Count errors by hour
def group_by_hour(errors):
    from collections import defaultdict
    hourly_counts = defaultdict(int)
 
    for error in errors:
        hour = error['timestamp'][:13]  # Extract hour portion
        hourly_counts[hour] += 1
 
    return hourly_counts
 
# Build pipeline
log_lines = read_log_file('app.log')
error_lines = filter_errors(log_lines)
parsed_errors = parse_error_lines(error_lines)
results = group_by_hour(parsed_errors)
 
print(results)

    이 파이프라인은 잠재적으로 매우 큰 로그 파일도 최소한의 메모리로 처리합니다. 최종 집계 단계에 도달하기 전까지는 어떤 시점에도 한 줄만 메모리에 존재합니다.

    데이터 변환 예시도 하나 더 보겠습니다:

    # Pipeline: numbers -> square -> filter evens -> sum
def square_numbers(numbers):
    for n in numbers:
        yield n * n
 
def filter_even(numbers):
    for n in numbers:
        if n % 2 == 0:
            yield n
 
# Chain the pipeline
numbers = range(1, 11)  # 1-10
squared = square_numbers(numbers)
evens = filter_even(squared)
result = sum(evens)  # Only even squares
 
print(result)  # 220 (4 + 16 + 36 + 64 + 100)

    메모리 비교: 제너레이터 vs 리스트 벤치마크

    제너레이터의 이점을 수치로 확인하기 위해, 실제에 가까운 메모리/성능 벤치마크를 해봅시다.

    import sys
import time
import tracemalloc
 
def process_with_list(n):
    """Traditional approach using lists"""
    tracemalloc.start()
    start_time = time.time()
 
    # Create list of squares
    squares = [x * x for x in range(n)]
 
    # Filter even squares
    even_squares = [x for x in squares if x % 2 == 0]
 
    # Sum results
    result = sum(even_squares)
 
    current, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    elapsed = time.time() - start_time
 
    return result, peak / 1024 / 1024, elapsed  # Convert to MB
 
def process_with_generator(n):
    """Generator approach"""
    tracemalloc.start()
    start_time = time.time()
 
    # Generator pipeline
    squares = (x * x for x in range(n))
    even_squares = (x for x in squares if x % 2 == 0)
    result = sum(even_squares)
 
    current, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    elapsed = time.time() - start_time
 
    return result, peak / 1024 / 1024, elapsed
 
# Benchmark with 1 million numbers
n = 1000000
 
list_result, list_memory, list_time = process_with_list(n)
gen_result, gen_memory, gen_time = process_with_generator(n)
 
print(f"Results match: {list_result == gen_result}")
print(f"\nList approach:")
print(f"  Memory: {list_memory:.2f} MB")
print(f"  Time: {list_time:.4f} seconds")
print(f"\nGenerator approach:")
print(f"  Memory: {gen_memory:.2f} MB")
print(f"  Time: {gen_time:.4f} seconds")
print(f"\nMemory savings: {((list_memory - gen_memory) / list_memory * 100):.1f}%")

    일반적인 출력:

    Results match: True

List approach:
  Memory: 36.21 MB
  Time: 0.0892 seconds

Generator approach:
  Memory: 0.12 MB
  Time: 0.0624 seconds

Memory savings: 99.7%

    제너레이터 방식은 99.7% 적은 메모리를 사용하면서 30% 더 빠르게 동작합니다. 데이터가 커질수록 이 차이는 더 크게 누적됩니다.

    itertools 모듈: 제너레이터 유틸리티

    Python의 itertools 모듈은 효율적인 반복을 위한 강력한 제너레이터 기반 도구 모음입니다. 이 유틸리티들은 C로 작성되어 매우 최적화되어 있습니다.

    필수 itertools 함수들

    import itertools
 
# chain - concatenate multiple iterables
combined = itertools.chain([1, 2], [3, 4], [5, 6])
print(list(combined))  # [1, 2, 3, 4, 5, 6]
 
# islice - slice an iterable (like list slicing but for generators)
numbers = itertools.count()  # Infinite counter: 0, 1, 2, 3...
first_ten = itertools.islice(numbers, 10)
print(list(first_ten))  # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
 
# count - infinite counter with start and step
counter = itertools.count(start=10, step=2)
print([next(counter) for _ in range(5)])  # [10, 12, 14, 16, 18]
 
# cycle - infinite repetition of an iterable
colors = itertools.cycle(['red', 'green', 'blue'])
print([next(colors) for _ in range(7)])  # ['red', 'green', 'blue', 'red', 'green', 'blue', 'red']
 
# accumulate - cumulative sums or other operations
numbers = [1, 2, 3, 4, 5]
cumulative = itertools.accumulate(numbers)
print(list(cumulative))  # [1, 3, 6, 10, 15]
 
# accumulate with custom function
import operator
products = itertools.accumulate(numbers, operator.mul)
print(list(products))  # [1, 2, 6, 24, 120]
 
# groupby - group consecutive elements by key
data = [('A', 1), ('A', 2), ('B', 3), ('B', 4), ('C', 5)]
for key, group in itertools.groupby(data, key=lambda x: x[0]):
    print(f"{key}: {list(group)}")
# A: [('A', 1), ('A', 2)]
# B: [('B', 3), ('B', 4)]
# C: [('C', 5)]

    실용적인 itertools 조합

    # Paginating results with islice
def paginate(iterable, page_size):
    iterator = iter(iterable)
    while True:
        page = list(itertools.islice(iterator, page_size))
        if not page:
            break
        yield page
 
# Usage
data = range(25)
for page_num, page in enumerate(paginate(data, 10), 1):
    print(f"Page {page_num}: {page}")
# Page 1: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
# Page 2: [10, 11, 12, 13, 14, 15, 16, 17, 18, 19]
# Page 3: [20, 21, 22, 23, 24]
 
# Windowed iteration (sliding window)
def window(iterable, size):
    it = iter(iterable)
    win = list(itertools.islice(it, size))
    if len(win) == size:
        yield tuple(win)
    for item in it:
        win = win[1:] + [item]
        yield tuple(win)
 
print(list(window([1, 2, 3, 4, 5], 3)))
# [(1, 2, 3), (2, 3, 4), (3, 4, 5)]

    실제 사용 사례

    대용량 파일을 한 줄씩 읽기

    def process_large_csv(filename):
    """Process a multi-GB CSV file efficiently"""
    with open(filename, 'r') as f:
        # Skip header
        next(f)
 
        for line in f:
            # Parse and yield record
            fields = line.strip().split(',')
            yield {
                'user_id': fields[0],
                'action': fields[1],
                'timestamp': fields[2]
            }
 
# Process millions of records with minimal memory
for record in process_large_csv('user_events.csv'):
    # Process one record at a time
    if record['action'] == 'purchase':
        print(f"Purchase by user {record['user_id']}")

    스트리밍 데이터 처리

    def stream_api_data(url, batch_size=100):
    """Stream paginated API data without loading all results"""
    offset = 0
 
    while True:
        response = requests.get(url, params={'offset': offset, 'limit': batch_size})
        data = response.json()
 
        if not data:
            break
 
        for item in data:
            yield item
 
        offset += batch_size
 
# Process unlimited API results
for item in stream_api_data('https://api.example.com/records'):
    process_item(item)

    데이터베이스 쿼리 결과 반복 처리

    def fetch_users_batch(cursor, batch_size=1000):
    """Fetch database records in batches without loading all into memory"""
    while True:
        results = cursor.fetchmany(batch_size)
        if not results:
            break
        for row in results:
            yield row
 
# Database query
cursor.execute("SELECT * FROM users WHERE active = 1")
 
# Process millions of users efficiently
for user in fetch_users_batch(cursor):
    send_email(user['email'], generate_report(user))

    ETL 파이프라인 예시

    # Extract: Read from source
def extract_from_csv(filename):
    with open(filename, 'r') as f:
        for line in f:
            yield line.strip().split(',')
 
# Transform: Clean and convert data
def transform_records(records):
    for record in records:
        yield {
            'id': int(record[0]),
            'name': record[1].title(),
            'email': record[2].lower(),
            'age': int(record[3]) if record[3] else None
        }
 
# Load: Write to database
def load_to_database(records, db_connection):
    for record in records:
        db_connection.execute(
            "INSERT INTO users VALUES (?, ?, ?, ?)",
            (record['id'], record['name'], record['email'], record['age'])
        )
        yield record  # Pass through for logging
 
# Build ETL pipeline
raw_data = extract_from_csv('users.csv')
transformed = transform_records(raw_data)
loaded = load_to_database(transformed, db_conn)
 
# Execute pipeline and count processed records
processed_count = sum(1 for _ in loaded)
print(f"Processed {processed_count} records")

    제너레이터 베스트 프랙티스와 흔한 함정

    베스트 프랙티스

    1. 단순한 경우에는 제너레이터 표현식을 사용

      # Simple transformation - use generator expression
squares = (x * x for x in range(1000))
 
# Complex logic - use generator function
def complex_processing(data):
    for item in data:
        # Multi-step processing
        result = step1(item)
        result = step2(result)
        if validate(result):
            yield result

    2. 데이터 파이프라인을 위해 제너레이터를 체이닝

      # Each stage processes lazily
data = read_source()
filtered = filter_stage(data)
transformed = transform_stage(filtered)
results = aggregate_stage(transformed)

    3. 위임에는 yield from을 사용

      def process_all_files(directory):
    for filename in os.listdir(directory):
        yield from process_file(filename)

    흔한 함정

    1. 제너레이터는 한 번 순회하면 소진됩니다

      gen = (x for x in range(3))
print(list(gen))  # [0, 1, 2]
print(list(gen))  # [] - exhausted!
 
# Solution: Convert to list or recreate generator
data = list(gen)  # If data fits in memory
# OR
gen = (x for x in range(3))  # Recreate

    2. 제너레이터는 len()이나 인덱싱을 지원하지 않습니다

      gen = (x for x in range(10))
# len(gen)  # TypeError
# gen[5]    # TypeError
 
# Solution: Convert to list if you need these operations
items = list(gen)
print(len(items))
print(items[5])

    3. 제너레이터 스코프/클로저에 주의

      # Wrong - all generators will use final value of i
generators = [lambda: i for i in range(3)]
print([g() for g in generators])  # [2, 2, 2]
 
# Correct - capture i in default argument
generators = [lambda i=i: i for i in range(3)]
print([g() for g in generators])  # [0, 1, 2]

    4. 제너레이터 체인에서의 예외 처리

      def stage1():
    for i in range(5):
        if i == 3:
            raise ValueError("Error in stage1")
        yield i
 
def stage2(data):
    try:
        for item in data:
            yield item * 2
    except ValueError as e:
        print(f"Caught: {e}")
        yield -1  # Error marker
 
# Exception is caught in stage2
for result in stage2(stage1()):
    print(result)

    비교: 제너레이터 vs 리스트 vs 이터레이터 vs map/filter

    FeatureGeneratorsListsIteratorsmap/filter
    Memory usageMinimal (lazy)Full datasetMinimal (lazy)Minimal (lazy)
    Creation speedInstantDepends on sizeInstantInstant
    ReusableNoYesNoNo
    IndexableNoYesNoNo
    len() supportNoYesNoNo
    ModificationRead-onlyMutableRead-onlyRead-only
    Infinite sequencesYesNoYesYes
    Syntaxyield or ()[]iter()map(), filter()
    Best forLarge datasets, pipelinesSmall datasets, random accessProtocol implementationFunctional transformations

    비교 예시:

    # All produce same results but with different characteristics
data = range(1000000)
 
# Generator - memory efficient, not reusable
gen = (x * 2 for x in data)
 
# List - memory intensive, reusable, indexable
lst = [x * 2 for x in data]
 
# map - memory efficient, functional style
mapped = map(lambda x: x * 2, data)
 
# Iterator - explicit protocol implementation
class Doubler:
    def __init__(self, data):
        self.data = iter(data)
 
    def __iter__(self):
        return self
 
    def __next__(self):
        return next(self.data) * 2
 
iterator = Doubler(data)

    Jupyter에서 제너레이터 실험하기

    제너레이터 패턴과 성능 특성을 탐구할 때는 인터랙티브 노트북 환경에서 작업하면 학습 속도가 빨라집니다. RunCell (opens in a new tab)은 AI 기반 보조 기능을 Jupyter 노트북에 직접 제공하므로, 제너레이터 기반 데이터 처리 파이프라인을 실험하는 데이터 사이언티스트에게 적합합니다.

    RunCell을 사용하면 다음을 할 수 있습니다:

    • 제너레이터 함수를 빠르게 프로토타이핑하고 메모리 특성을 테스트
    • 실제 데이터셋으로 제너레이터 vs 리스트 성능 벤치마크
    • 복잡한 제너레이터 파이프라인을 인터랙티브하게 구성하고 디버깅
    • 제너레이터 기반 ETL 워크플로우 최적화를 위한 AI 제안 받기

    노트북에서 제너레이터를 탐색하는 예시는 다음과 같습니다:

    # Cell 1: Define generator pipeline
def read_data():
    for i in range(1000000):
        yield {'id': i, 'value': i * 2}
 
def filter_large(records):
    for record in records:
        if record['value'] > 1000:
            yield record
 
def transform(records):
    for record in records:
        record['squared'] = record['value'] ** 2
        yield record
 
# Cell 2: Execute pipeline and measure
import time
start = time.time()
 
pipeline = transform(filter_large(read_data()))
results = list(itertools.islice(pipeline, 100))  # Take first 100
 
print(f"Time: {time.time() - start:.4f}s")
print(f"Results: {len(results)}")
 
# Cell 3: Visualize with PyGWalker
import pygwalker as pyg
pyg.walk(results)

    FAQ

    결론

    Python 제너레이터는 즉시 평가(eager evaluation)에서 지연 평가(lazy evaluation)로의 근본적인 전환을 의미하며, 수천 개에서 수십억 개 레코드에 이르는 데이터셋을 메모리 효율적으로 처리할 수 있게 해줍니다. yield, 제너레이터 표현식, 이터레이터 프로토콜, 그리고 send(), yield from 같은 고급 기능을 이해하면, 손쉽게 확장 가능한 정교한 데이터 처리 파이프라인을 만들 수 있습니다.

    기억해야 할 핵심 인사이트는 다음과 같습니다.

    • 제너레이터는 지연 평가를 통해 메모리 사용량을 최소화하며, 종종 리스트 대비 99%+ 절감 효과가 있습니다.
    • 단순 변환에는 제너레이터 표현식을, 복잡한 로직에는 제너레이터 함수를 사용하세요.
    • 제너레이터를 체이닝해 메모리 효율적인 데이터 처리 파이프라인을 구성하세요.
    • 강력한 제너레이터 기반 반복 유틸리티를 위해 itertools를 활용하세요.
    • 대규모 데이터와 단일 패스 반복에는 제너레이터를, 랜덤 액세스가 필요한 소규모 데이터에는 리스트를 선택하세요.

    대용량 로그 파일 처리, API 데이터 스트리밍, ETL 파이프라인 구축 등 어떤 경우든 제너레이터는 프로덕션 규모의 데이터 처리에 필요한 성능과 메모리 효율을 제공합니다. 이 패턴들을 익히면 어떤 크기의 데이터셋도 우아하고 효율적으로 다루는 Python 코드를 작성할 수 있게 됩니다.

    📚