Python 生成器：提升性能和简化代码

什么是Python生成器？

在本质上，Python 生成器是一种特殊类型的函数，甚至是一种产生值序列的紧凑表达式，它们惰性生成值。把生成器想象成工厂中的传送带：不是把所有产品堆积在一个地方并且空间不足，而是在产品下线时逐个处理。这使得生成器在内存上更有效率，并且是 Python 的迭代器协议的自然延伸，该协议支持 Python 的诸多内置工具，如 for 循环和推导式。

生成器背后的魔法在于 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生成器与迭代器

Python中的传统迭代器需要具有显式__iter__()和__next__()方法的类，这涉及大量样板代码和手动状态管理，而生成器函数简化了这一过程，通过简单的函数演示了每个数字的平方值直到n。

为什么我们使用Python生成器

在解释什么是Python生成器时，我也传达了一些它们为什么被使用的想法。在这一部分，我想更详细地探讨一下。因为生成器不仅仅是Python的一个花哨特性，而是真正解决实际问题的。

内存效率  

与存储所有元素的列表或数组不同，生成器会动态产生值，因此一次只在内存中保存一个项目。

例如，考虑Python 2的range()和xrange()之间的区别：

• range()在内存中创建了一个列表，对于大范围可能会有问题。

• xrange()像生成器一样工作，惰性地产生值。

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

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

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()方法允许您将值传递回生成器，将其转换为协程。这对于创建交互式或有状态的生成器非常有用。

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()

.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()

方法通过引发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))

内存和速度基准

由于生成器的工作方式，它们在对内存效率至关重要的场景中表现出色，但（您可能会感到惊讶）它们并不总是最快的选项。让我们将生成器与列表进行比较，以了解它们的权衡之处。

之前，我们展示了生成器在内存方面优于列表。这是我们比较生成10百万个数字序列时的内存使用情况。现在让我们来做一件不同的事情，进行速度比较：

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编程技能路径。点击下方链接开始。