Python 生成器：提升性能並簡化程式碼

什麼是Python生成器？

在核心，Python 生成器是一种特殊类型的函数，甚至是一个产生值序列的紧凑表达式，以惰性方式生成值。将生成器看作是工厂中的传送带：不是将所有产品堆放在一个地方并耗尽空间，而是在每个项目下来时处理它。这使生成器在内存上更有效，并且是Python的迭代器协议的自然扩展，该协议支持诸如for循环和推导式之类的Python内置工具。

生成器背后的魔法在于yield关键字。与return不同，后者输出单个值并退出函数，yield生成一个值，暂停函数的执行，并保存其状态。当再次调用生成器时，它会从停下的地方继续执行。

例如，想象一下你正在逐行读取一个庞大的日志文件。生成器可以逐行处理每行而无需将整个文件加载到内存中。这种“惰性评估”使生成器与传统函数有所不同，并使它们成为性能敏感任务的首选工具。

一个基本的Python生成器示例

讓我們練習一下，以熟悉這個概念。這裡是一個產生前n個整數的生成函數。

def generate_integers(n):
    for i in range(n):
        yield i  # 暫停並返回 i

# 使用生成器
for num in generate_integers(5):
    print(num)

0
1
2
3
4

我創建了一個視覺圖，幫助您看到底層發生了什麼：

Python 生成器語法和模式

生成器可以以多種方式實現。也就是說，有兩種主要方式：生成函數和生成表達式。

生成函數

生成函數的定義類似於常規函數，但使用yield關鍵字而不是return。調用時，它返回一個可以迭代的生成器物件。

def count_up_to(n):
    count = 1
    while count <= n:
        yield count
        count += 1

# 使用生成器
counter = count_up_to(5)
for num in counter:
    print(num)

1
2
3
4
5

從上述範例中，我們可以看到當函數 count_up_to 被調用時，它會返回一個生成器對象。每當 for 循環請求一個值時，函數會運行直到碰到 yield，產生當前的 count 值並在迭代之間保留其狀態，以便可以精確地從中斷的地方繼續。

生成器表達式

生成器表達式是一種創建生成器的簡潔方式。它們類似於列表推導式，但使用括號而不是方括號。

# 列表推導式（即時評估）
squares_list = [x**2 for x in range(5)]  # [0, 1, 4, 9, 16]

# 生成器表達式（延遲評估）
squares_gen = (x**2 for x in range(5))

# 使用生成器
for square in squares_gen:
    print(square)

0
1
4
9
16

那麼，列表推導和生成器表達式之間有什麼區別？列表推導會在記憶體中創建整個列表，而生成器表達式則一次生成一個值，節省記憶體。如果您對列表推導不熟悉，可以閱讀我們的《Python列表推導教程》。Python生成器 vs. 迭代器

Python中的傳統迭代器需要具有明確__iter__()和__next__()方法的類，這涉及大量樣板代碼和手動狀態管理，而生成器函數通過自動保存狀態並消除這些方法的需要來簡化過程——通過一個簡單函數的示例，該函數產生每個數字的平方直到n。

我們為什麼使用Python生成器

在解釋什麼是Python生成器時，我也傳達了一些它們被使用的原因。在這一部分，我想更詳細地探討一下。因為生成器不僅是一個花俏的Python特性，而且它們確實解決了真實的問題。

內存效率  

與列表或數組不同，它們同時將所有元素存儲在內存中，生成器可以即時產生值，因此一次只在內存中保存一個項目。  

例如，考慮Python 2中range()和xrange()之間的區別：  

•  range()在內存中創建了一個列表，對於大範圍可能會出現問題。

• xrange() 行为像一个生成器，惰性产生值。

由于 xrange() 的行为更有用，现在在 Python 3 中，range() 也像生成器一样运行，因此避免了同时存储所有值的内存开销。

为了展示这个概念，让我们比较生成一个一千万个数的序列时的内存使用情况：

import sys

# 使用列表
numbers_list = [x for x in range(10_000_000)]
print(f"Memory used by list: {sys.getsizeof(numbers_list) / 1_000_000:.2f} MB")

# 使用生成器
numbers_gen = (x for x in range(10_000_000))
print(f"Memory used by generator: {sys.getsizeof(numbers_gen)} bytes")

Memory used by list: 89.48 MB
Memory used by the generator: 112 bytes

正如您所看到的，與列表相比，生成器幾乎不佔用內存，這種差異是顯著的。

性能改進 

由於惰性求值，值僅在需要時計算。這意味著您可以立即開始處理數據，而無需等待整個序列生成完成。

例如，想象一下計算前100萬個數字的平方和：

# 使用列表（急切求值）
sum_of_squares_list = sum([x**2 for x in range(1_000_000)])

# 使用生成器（惰性求值）
sum_of_squares_gen = sum(x**2 for x in range(1_000_000))

雖然兩種方法都會給出相同的結果，但生成器版本避免了創建一個龐大的列表，因此我們可以更快地獲得結果。

簡單和可讀性

生成器通過消除樣板代碼簡化了迭代器的實現。比較基於類的迭代器和生成器函數：

這是基於類的迭代器：

class SquaresIterator:
    def __init__(self, n):
        self.n = n
        self.current = 0
    def __iter__(self):
        return self
    def __next__(self):
        if self.current >= self.n:
            raise StopIteration
        result = self.current ** 2
        self.current += 1
        return result

# 用法
squares = SquaresIterator(5)
for square in squares:
    print(square)

這是生成器函數：

def squares_generator(n):
    for i in range(n):
        yield i ** 2

# 用法
squares = squares_generator(5)
for square in squares:
    print(square)

生成器版本更短、更易讀，且不需要樣板代碼。這是 Python 哲學的完美例證：簡單就是更好。

處理無限序列

最後，我想說生成器非常適合表示無限序列，這是列表根本無法做到的。例如，考慮 費波那契數列：

def fibonacci():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

# 用法
fib = fibonacci()
for _ in range(10):
    print(next(fib))

0
1
1
2
3
5
8
13
21
34

這個生成器可以無限地產生費波那契數列，而且不會耗盡記憶體。其他例子包括處理即時數據流或處理時間序列數據。

高級Python生成器概念  

現在，讓我們來看一些更難的概念。在這一部分中，我們將探索如何組合生成器，並使用獨特的生成器方法，如.send()、.throw()和.close()。  

將生成器串連在一起

生成器可以結合在一起。您可以通過將生成器串連在一起，以模塊化地轉換、過濾和處理數據。 

假設您有一個無限數字序列，並想要對每個數字進行平方並過濾出奇數結果：

def infinite_sequence():
    num = 0
    while True:
        yield num
        num += 1
def square_numbers(sequence):
    for num in sequence:
        yield num ** 2
def filter_evens(sequence):
    for num in sequence:
        if num % 2 == 0:
            yield num

# 組合生成器
numbers = infinite_sequence()
squared = square_numbers(numbers)
evens = filter_evens(squared)

# 輸出前10個偶數的平方
for _ in range(10):
    print(next(evens))

0
4
16
36
64
100
144
196
256
324

該過程涉及infinite_sequence函數無限生成數字，而square_numbers生成每個數字的平方，然後filter_evens過濾出奇數，僅生成偶數的平方。我們的Python 開發人員職業軌道涉及這種類型的事情，因此您可以看到如何使用生成器構建和調試複雜的管道，以及迭代器和列表理解。

特殊的生成器方法

生成器配備了高級方法，允許雙向通信和控制終止。

send()

.send()方法允許您將值傳回生成器，將其轉換為協程。這對於創建交互式或具有狀態的生成器非常有用。

def accumulator():
    total = 0
    while True:
        value = yield total
        if value is not None:
            total += value

# 使用生成器
acc = accumulator()
next(acc)  # 啟動生成器
print(acc.send(10))  # 輸出：10
print(acc.send(5))   # 輸出：15
print(acc.send(20))  # 輸出：35

這是它的運作方式：

• 生成器開始使用next(acc)進行初始化。

• 每次呼叫.send(value)都會將一個值傳入生成器，該值將被分配給yield語句中的value。

• 生成器更新其状态（total）并产生新的结果。

.throw()方法允许您在生成器内部引发异常，这对于错误处理或信号特定条件很有帮助。

def resilient_generator():
    try:
        for i in range(5):
            yield i
    except ValueError:
        yield "Error occurred!"

# 使用生成器
gen = resilient_generator()
print(next(gen))  # 输出: 0
print(next(gen))  # 输出: 1
print(gen.throw(ValueError))  # 输出: "发生错误！"

這是它的工作原理：

• 生成器通常運行直到調用.throw()。

• 當例外在生成器內部引發時，可以使用try-except塊處理它。

close()

使用 .close() 方法可以通過引發 GeneratorExit 異常來停止生成器。這對於清理資源或停止無限生成器非常有用。

def infinite_counter():
    count = 0
    try:
        while True:
            yield count
            count += 1
    except GeneratorExit:
        print("Generator closed!")

# 使用生成器
counter = infinite_counter()

print(next(counter))  # 輸出：0
print(next(counter))  # 輸出：1
counter.close()       # 輸出："生成器已關閉！"

這是它的工作原理：

• 生成器運行直到調用 .close()。

• 生成器引發GeneratorExit例外，允許生成器在終止之前進行清理或記錄消息。

數據科學中的實際應用  

希望您開始欣賞生成器是多麼有用。在本節中，我將試圖突出使用案例，讓您能夠想像它們在您日常工作中的實際應用。

處理大型數據集  

在數據科學中最常見的挑戰之一是處理無法容納在內存中的大型數據集。生成器提供了逐行處理此類數據的方法。  

假設您有一個包含銷售數據的 10 GB CSV 檔案，需要篩選特定地區的記錄。以下是您可以使用生成器管道來實現此目的的方法：

import csv
def read_large_csv(file_path):
    """ Generator to read a large CSV file line by line."""
    with open(file_path, mode="r") as file:
        reader = csv.DictReader(file)
        for row in reader:
            yield row

def filter_by_region(data, region):
    """ Generator to filter rows by a specific region."""
    for row in data:
        if row["Region"] == region:
            yield row

# 生成器管道
file_path = "sales_data.csv"
region = "North America"
data = read_large_csv(file_path)
filtered_data = filter_by_region(data, region)

# 處理篩選後的數據
for record in filtered_data:
    print(record)

這裡是正在發生的事情：

1. read_large_csv 逐行讀取文件，將每一行作為字典逐個生成。

2. filter_by_region 根據指定的地區篩選行。

3. 管道會以增量方式處理數據，避免記憶體超載。

這種方法有利於提取、轉換和加載工作流程，在分析之前必須先清理和轉換數據。您將在我們的 Python ETL 和 ELT 課程中看到這種情況。流式處理和管道

有時數據會以連續流的方式到達。想想傳感器數據、實時信息流或社交媒體。

假設您正在使用每秒生成溫度讀數的物聯網設備。您希望計算在10個讀數的滑動窗口上的平均溫度：

def sensor_data_stream():
    """Simulate an infinite stream of sensor data."""
    import random
    while True:
        yield random.uniform(0, 100)  # 模擬感應器數據

def sliding_window_average(stream, window_size):
    """ Calculate the average over a sliding window of readings."""
    window = []
    for value in stream:
        window.append(value)
        if len(window) > window_size:
            window.pop(0)
        if len(window) == window_size:
            yield sum(window) / window_size

# 生成器流水線
sensor_stream = sensor_data_stream()
averages = sliding_window_average(sensor_stream, window_size=10)

# 每秒打印平均值
for avg in averages:
    print(f"Average temperature: {avg:.2f}")

這是解釋：

1. sensor_data_stream 模擬無限的感應器讀數流。

2. sliding_window_average 維持著最後 10 次讀數的滑動窗口並計算它們的平均值。

3. 該管道實時處理數據，非常適合監控和分析。

其他使用案例

生成器也用於數據大小不可預測或數據不斷流入/無限的情況。

網頁爬蟲

當爬取網站時，您通常不知道需要處理多少頁面或項目。生成器允許您優雅地處理這種不可預測性：

def scrape_website(url):
    """ Generator to scrape a website page by page."""
    while url:
        # 模擬獲取和解析頁面
        print(f"Scraping {url}")
        data = f"Data from {url}"
        yield data
        url = get_next_page(url)  # 假想的獲取下一頁的函數

# 用法
scraper = scrape_website("https://example.com/page1")
for data in scraper:
    print(data)

模擬任務

在模擬中，比如蒙特卡洛方法或遊戲開發，生成器可以表示無限或動態序列：

def monte_carlo_simulation():
    """ Generator to simulate random events for Monte Carlo analysis."""
    import random
    while True:
        yield random.random()

# 用法
simulation = monte_carlo_simulation()
for _ in range(10):
    print(next(simulation))

內存和速度基準

由於它們的工作方式，生成器在內存效率至關重要的情況下表現出色，但（您可能會感到驚訝）它們並不總是最快的選擇。讓我們將生成器與列表進行比較，以了解它們之間的權衡。

之前，我們展示了生成器在內存方面比列表更優越。這是我們比較生成一千萬個數字序列時的內存使用情況的部分。現在讓我們做一個不同的事情，進行速度比較：

import time

# 列表推導
start_time = time.time()
sum([x**2 for x in range(1_000_000)])
print(f"List comprehension time: {time.time() - start_time:.4f} seconds")

# 生成器表達式
start_time = time.time()
sum(x**2 for x in range(1_000_000))
print(f"Generator expression time: {time.time() - start_time:.4f} seconds")

List comprehension time: 0.1234 seconds
Generator expression time: 0.1456 seconds

雖然生成器節省了內存，但在這種情況下，它實際上比列表慢。這是因為對於這個較小的數據集，存在暫停和恢復執行的開銷。

對於小型數據集，性能差異微不足道，但對於大型數據集，生成器的內存節省往往超過了輕微的速度損失。

出現的問題

最後，讓我們看看一些常見的錯誤或問題：

生成器是耗盡的

一旦生成器耗盡，就無法再次使用。如果您想要再次迭代，您需要重新創建它。

gen = (x for x in range(5))
print(list(gen))  # 輸出: [0, 1, 2, 3, 4]
print(list(gen))  # 輸出: [] (生成器已耗盡)

延遲評估可能會很棘手

由於生成器按需生成值，因此錯誤或副作用可能直到生成器被迭代時才會出現。

您可能過度使用生成器

對於小型數據集或簡單任務，使用生成器的開銷可能不值得節省內存。考慮這個例子，我正在為多次迭代生成數據。

# 生成器表達式
gen = (x**2 for x in range(10))

# 轉換為列表
squares = list(gen)

# 重用該列表
print(sum(squares))  # 輸出: 285
print(max(squares))  # 輸出: 81

選擇何時使用生成器

總結來說，我將提供一些非常一般的規則，說明何時使用生成器。使用於：

• 大型數據集：當處理的數據集過大無法放入記憶體時，使用生成器。
• 無限序列：使用生成器來表示無限序列，例如實時數據流或模擬。
• 管道：使用生成器來構建模組化的數據處理管道，逐步轉換和過濾數據。

何時應該將資料實體化（轉換為列表）

• 小型資料集：如果記憶體不是問題且需要快速訪問所有元素，請不要使用生成器；請改用列表。
• 多次迭代：如果需要多次迭代相同的資料，請不要使用生成器；請將其實體化為列表，以避免重複創建生成器。

結論和重點摘要

在這篇文章中，我們探討了生成器如何幫助你應對數據科學中的現實挑戰，從處理大型數據集到構建實時數據管道。繼續練習。掌握生成器的最佳方法是將其應用於你自己的工作中。作為開始，嘗試用生成器表達式替換列表推導，或將循環重構為生成器函數。

一旦你掌握了基礎知識，就可以探索建立在生成器概念上的新主題和更高級的主題：

• 協程：使用 .send() 和 .throw() 創建可以接收和處理數據的生成器，實現雙向通信。

• 非同步程式設計：將生成器與 Python 的 asyncio 函式庫結合，以構建高效的非阻塞應用程式。  

• 併發：學習生成器如何實現協作式多任務處理和輕量級併發。  

持續學習並成為專家。立即參加我們的 Python 開發者 職業路徑或我們的 Python 程式設計 技能路徑。點擊下面的鏈接開始吧。