如何低成本提高Python运行速度

本文介绍了三种提高Python类内存效率的技术和方法。通过遵循这些建议，你可以优化类的内存使用，从而提升整体性能。无论是处理数据密集型项目还是面向对象编程，创建高效利用内存的类都至关重要，值得我们关注和实践。

1. 使用__slots__

使用 Python 的 __slots__ 可以显式地定义类可以拥有的属性。这通常可以避免创建动态字典来存储属性，从而优化类的内存使用。

Python 默认情况下将实例属性存储在私有字典 __dict__中。这个字典允许很大的灵活性，允许运行时添加、修改或删除属性。然而，这种灵活性通常是以内存开销为代价的。类的每个实例都有一个字典，以键值对的形式存储属性名和值。使用 __slots__时，Python 直接为每个实例中的指定属性保留固定的空间，而不是使用默认的字典。

下面是一个使用 __slots__ 来提高内存效率的 Python 类的示例：

class Ant:
    __slots__ = ['worker_id', 'role', 'colony']

    def __init__(self, worker_id, role, colony):
        self.worker_id = worker_id
        self.role = role
        self.colony = colony

# Instantiate multiple Ant objects
ant0 = Ant("Q", "Queen", "Red Colony")
ant1 = Ant("W1", "Worker", "Red Colony")
ant2 = Ant("W2", "Worker", "Red Colony")
ant3 = Ant("S1", "Soldier", "Red Colony")

在本例中，Ant 类使用 __slots__明确定义了worker_id、role和colony属性。这种特殊性避免了为属性存储创建动态字典，从而在创建多个 Ant 类实例时节省了内存。

当需要创建一个类的大量实例时（如创建一个蚁群时），使用 __slots__ 的好处会变得更加显著。如果没有 __slots__，使用属性字典（python 的默认设置）的开销就会变得很大，导致内存使用量增加，性能也可能下降。

一个包含蚂蚁成员列表的 Colony 类，如下所示：

class Colony:
    def __init__(self, name):
 self.name = name
        self.ants = []

    def add_ant(self, worker_id, role):
        ant = Ant(worker_id, role, self.name)
        self.ants.append(ant)

    def distribute_work(self):
        # add code to distribute work among the ants
        pass

    def defend_queen(self):
        # add code to defend the queen
        pass

实例化一个蚁群，然后运行一个循环，向实例中添加 500 000 只蚂蚁：

# Create an instance of Colony
colony_name = "Tinyopolis"
colony = Colony(colony_name)

# Simulate an ant colony of 500,000 worker ants
n_ants = 500_000
for i in range(n_ants):
    worker_id = f"W{i}"
    role = "Worker"
    colony.add_ant(worker_id, role)

当我们反复实例化 Ant() 类时，使用 __slots__ 可以减少内存占用。使用 pympler 软件包剖析这个循环的内存使用情况，可以验证这一事实。比较使用 __slots__ 和不使用 __slots__ 的类的每次迭代的内存使用量时，我们得到以下结果：

在这里可以看到，使用 __slots__所占用的内存只有传统定义的类（默认使用 __dict__）的一半左右。

__slots__限制了可以分配给实例的属性，只有 __slots__中列出的属性才能直接分配和访问实例。任何分配未列在 __slots__中的属性的尝试都会引发 AttributeError。这有助于防止因输入错误而意外创建属性，但如果在开发后期需要添加其他属性，这也会造成限制。

__slots__ 可以通过消除对每个实例字典的需求，提高内存效率，使对象更紧凑， 减少总体内存使用。在创建大量类实例时尤其有用，有助于优化内存消耗和提高整体性能。此外，还可以从更快的属性访问时间中受益，与具体使用情况相关。

2. 使用惰性初始化

惰性初始化（Lazy Initialization）惰性初始化是一种延迟加载的策略，意味着只有在真正需要对象时才进行初始化。这种策略通常用于优化性能和资源使用，特别是在对象创建成本较高或资源有限的情况下。

在Python中，可以使用functools.cached_property装饰器实现惰性初始化。这个装饰器允许定义只计算一次的属性，并缓存起来，以便以后访问。通过使用@cached_property装饰器，在首次访问数据集时可以惰性加载数据集，而不是提前加载。

下面的示例说明了如何使用 cached_property 在 Python 类中惰性地加载数据集：

from functools import cached_property

class DataLoader:

    def __init__(self, path):
        self.path = path

    @cached_property
    def dataset(self):
        # 在此加载数据集
        # 这只会在首次访问数据集属性时执行一次
        return self._load_dataset()

    def _load_dataset(self):
        print("Loading the dataset...")

        # load a big dataset here
        df = pd.read_csv(self.path)
        return df

# instantiate the DataLoader class
path = "/[path_to_dataset]/mnist.csv"
mnist = DataLoader(path)

在这个例子中，DataLoader 类通过 cached_property 装饰器定义了一个 dataset 属性。_load_dataset 方法负责首次访问 dataset 属性时的数据集加载。后续访问 dataset 属性将返回缓存值，而不会重新加载数据集。

对于处理大型数据集时，这种惰性初始化方法非常有用。在这个例子中，我将展示通过 DataLoader 类加载 MNIST 数据集，并比较在访问 dataset 属性前后的内存占用情况。尽管 MNIST 数据集本身并不是很大，但它有效地说明了我的观点。

在实际例子中，考虑在庞大数据集上执行复杂处理步骤的 DataProcessor 类。可以使用 DataLoader 类，该类可以懒散地加载数据并利用 cached_property 装饰器。这种方法允许在调用特定方法时加载数据集，从而按需进行数据处理，节省内存并提高性能。以下是一个实现示例：

class DataProcessor:
    def __init__(self, path):
 self.path = path
        self.data_loader = DataLoader(self.path)

    def process_data(self):
        dataset = self.data_loader.dataset
        print("Processing the dataset...")
        # 对加载的数据集执行复杂的数据处理步骤
        ...
    
# instantiate the DataLoader class
path = "/[path_to_dataset]/mnist.csv"

# 使用数据文件路径实例化 DataProcessor 类
# 此阶段不会加载数据！✅
processor = DataProcessor(path)

# 触发处理
processor.process_data()  # 数据集将在需要时加载和处理

到目前为止，一切顺利。但如果数据集非常大，无法一次装入内存怎么办？现在，懒散地加载数据集并不一定有帮助，我们需要想其他办法来保证类的内存效率。

3. 使用生成器

Python生成器是一种可迭代类型，类似于列表和元组，但有一个关键区别。生成器不会将所有值一次性存储在内存中，而是在需要时即时生成值。这使得生成器在处理大量数据时具有很高的内存效率。

在处理大型数据集时，生成器特别有用。生成器允许你一次生成或加载一个数据块，这有助于节省内存。这种方法为按需处理和迭代大量数据提供了一种更有效的方式。

下面是一个 ChunkProcessor 类的示例，该类使用生成器分块加载数据、处理数据并将数据保存到另一个文件中：

import pandas as pd

class ChunkProcessor:
    def __init__(self, filepath, chunk_size, verbose=True):
        self.filepath = filepath
        self.chunk_size = chunk_size
        self.verbose = verbose

    def process_data(self):
        for chunk_id, chunk in enumerate(self.load_data()):
            processed_chunk = self.process_chunk(chunk)
            self.save_chunk(processed_chunk, chunk_id)

    def load_data(self):
        # load data in chunks
        skip_rows = 0
        while True:
            chunk = pd.read_csv(self.filepath, skiprows=skip_rows, nrows=self.chunk_size)
            if chunk.empty:
                break
            skip_rows += self.chunk_size
            yield chunk

    def process_chunk(self, chunk):
        # process each chunk of data
        processed_chunk = processing_function(chunk)
        return processed_chunk

    def save_chunk(self, chunk, chunk_id):
        # save each processed chunk to a parquet file
        chunk_filepath = f"./output_chunk_{chunk_id}.parquet"
        chunk.to_parquet(chunk_filepath)
        if self.verbose:
            print(f"saved {chunk_filepath}")

在DataProcessor类中，load_data方法使用yield关键字来分块读取数据集，使其成为一个生成器。这样，它可以分块加载数据，并在加载下一个数据块时丢弃每个数据块。process_data方法对生成器进行迭代，以数据块为单位处理数据，并将每个数据块保存为单独的文件。

虽然 load_data 方法可以高效处理和迭代大型数据集，但它有限制。该实现仅支持加载保存在磁盘上的 CSV 文件，无法以相同方式加载 Parquet 文件，因为它们以列为单位的格式存储，不支持跳行。但如果 Parquet 文件已分块保存在磁盘上，则可以进行分块加载。因此，为了提高性能，我们会将最终处理好的文件保存为分块的 Parquet 格式，避免未来需要重新分解的麻烦。

如果使用 pandas 加载 CSV 文件，可以在 pd.read_csv() 中使用 chunksize 参数来节省时间和代码。该参数会自动返回一个生成器，因此无需在 load_data()中编写所有模板代码。下面是使用 pandas 实现的简化代码：

import pandas as pd

class PandasChunkProcessor:
    def __init__(self, filepath, chunk_size, verbose=True):
        self.filepath = filepath
        self.chunk_size = chunk_size
        self.verbose = verbose

    def process_data(self):
        for chunk_id, chunk in enumerate(pd.read_csv(self.filepath, chunksize=self.chunk_size)):
            processed_chunk = self.process_chunk(chunk)
            self.save_chunk(processed_chunk, chunk_id)

    def process_chunk(self, chunk):
        # process each chunk of data
        processed_chunk = processing_function(chunk)
        return processed_chunk

    def save_chunk(self, chunk, chunk_id):
        # save each processed chunk to a parquet file
        chunk_filepath = f"./output_chunk_{chunk_id}.parquet"
        chunk.to_parquet(chunk_filepath)
        if self.verbose:
            print(f"saved {chunk_filepath}")

使用生成器来节省内存的另一个注意事项是，并行处理生成器并不像 Python 中的列表那样简单。如果你的数据足够大，需要并行处理，你可能不得不考虑使用 concurrent.futures 或

本文范围之外的其他高级技术