Python高级技术之：如何利用`Cython`为`Pandas`和`NumPy`编写高性能的`C`扩展。

各位久等了，今天咱们聊聊如何用 Cython 这把瑞士军刀，给 Pandas 和 NumPy 这俩数据分析界的扛把子，打造高性能的 C 扩展，让他们如虎添翼，跑得更快！

开场白：Python 的速度，永远的痛？

Python 易学易用，库也多如繁星，但在处理大规模数据和复杂计算时，速度就成了绕不开的坎儿。尤其是 Pandas 和 NumPy，虽然它们本身已经做了很多优化，但遇到性能瓶颈时，还是得另辟蹊径。这时候，Cython 就该闪亮登场了。

Cython 是什么？一句话概括：Python + C 的混血儿

Cython 是一种编程语言，它是 Python 的超集，允许你编写 C 代码，并能无缝地与 Python 代码集成。换句话说，你可以用 Cython 来编写高性能的 C 扩展，然后在 Python 中像调用普通 Python 模块一样调用它们。

为什么选择 Cython？

• 性能提升： C 代码的执行速度远快于 Python 代码，尤其是对于循环和数值计算。
• 代码复用： 可以利用现有的 C/C++ 代码库。
• 易于集成： 可以轻松地将 C 扩展集成到 Python 项目中。
• 相对简单： 相比直接编写 C 扩展，Cython 的学习曲线更平缓。

准备工作：磨刀不误砍柴工

1. 安装 Cython：

   pip install cython

2. 安装 NumPy 和 Pandas (如果没有的话):

   pip install numpy pandas

3. 安装 C 编译器：

   • Windows： 需要安装 Visual Studio 或 MinGW。
   • Linux/macOS： 通常自带 GCC。

实战演练：给 Pandas 加速

咱们先从一个简单的例子入手，看看如何用 Cython 加速 Pandas 的操作。

场景：计算 DataFrame 中每一行的平均值

假设我们有一个 DataFrame，包含大量的数值数据，我们需要计算每一行的平均值。

1. Python (Pandas) 实现：

import pandas as pd
import numpy as np

def calculate_row_mean_python(df):
    """
    使用 Pandas 计算 DataFrame 中每一行的平均值。
    """
    return df.mean(axis=1)

# 创建一个示例 DataFrame
data = np.random.rand(1000, 100)  # 1000 行，100 列
df = pd.DataFrame(data)

# 测试 Python 实现
result_python = calculate_row_mean_python(df)
print(result_python)

2. Cython 实现：

• 创建一个名为 row_mean.pyx 的文件，内容如下：

# cython: language_level=3
import numpy as np
cimport numpy as cnp
import pandas as pd

def calculate_row_mean_cython(df: pd.DataFrame):
    """
    使用 Cython 计算 DataFrame 中每一行的平均值。
    """
    cdef int nrows = len(df)
    cdef int ncols = len(df.columns)
    cdef cnp.ndarray[cnp.float64_t, ndim=1] means = np.zeros(nrows, dtype=np.float64)
    cdef cnp.ndarray[cnp.float64_t, ndim=2] data = df.values

    for i in range(nrows):
        cdef double sum = 0.0
        for j in range(ncols):
            sum += data[i, j]
        means[i] = sum / ncols

    return means

• 创建一个名为 setup.py 的文件，用于编译 Cython 代码：

from setuptools import setup
from Cython.Build import cythonize
import numpy

setup(
    ext_modules = cythonize("row_mean.pyx"),
    include_dirs=[numpy.get_include()]
)

• 编译 Cython 代码：

python setup.py build_ext --inplace

• 在 Python 中调用 Cython 代码：

import pandas as pd
import numpy as np
import row_mean  # 导入编译后的 Cython 模块

def calculate_row_mean_python(df):
    """
    使用 Pandas 计算 DataFrame 中每一行的平均值。
    """
    return df.mean(axis=1)

# 创建一个示例 DataFrame
data = np.random.rand(1000, 100)  # 1000 行，100 列
df = pd.DataFrame(data)

# 测试 Python 实现
result_python = calculate_row_mean_python(df)

# 测试 Cython 实现
result_cython = row_mean.calculate_row_mean_cython(df)

# 验证结果是否一致
np.testing.assert_allclose(result_python.values, result_cython)

print("结果一致")

代码解释：

• row_mean.pyx:
  • # cython: language_level=3: 指定 Cython 编译器的语言级别为 Python 3。
  • import numpy as np: 导入 NumPy 库。
  • cimport numpy as cnp: 使用 cimport 导入 NumPy 的 C 定义，以便在 Cython 代码中使用 NumPy 的 C API。
  • import pandas as pd: 导入 Pandas 库。
  • def calculate_row_mean_cython(df: pd.DataFrame):: 定义一个函数，接受一个 Pandas DataFrame 作为输入。类型提示 pd.DataFrame 告诉 Cython df 是一个 Pandas DataFrame，虽然Cython在编译时主要关注变量的C类型信息，但这些类型提示有助于提高代码的可读性，并帮助静态分析工具进行类型检查。
  • cdef int nrows = len(df): 获取 DataFrame 的行数，并将其声明为 C 类型的整数 (cdef int)。
  • cdef int ncols = len(df.columns): 获取 DataFrame 的列数，并将其声明为 C 类型的整数。
  • cdef cnp.ndarray[cnp.float64_t, ndim=1] means = np.zeros(nrows, dtype=np.float64): 创建一个 NumPy 数组，用于存储每一行的平均值。cnp.ndarray[cnp.float64_t, ndim=1] 指定数组的类型为 float64，维度为 1。
  • cdef cnp.ndarray[cnp.float64_t, ndim=2] data = df.values: 将 DataFrame 的数据转换为 NumPy 数组，并将其声明为 C 类型的二维数组。cnp.ndarray[cnp.float64_t, ndim=2] 指定数组的类型为 float64，维度为 2。
  • for i in range(nrows):: 使用 C 风格的 for 循环遍历每一行。
  • cdef double sum = 0.0: 声明一个 C 类型的双精度浮点数，用于存储每一行的和。
  • for j in range(ncols):: 使用 C 风格的 for 循环遍历每一列。
  • sum += data[i, j]: 将当前元素的值加到 sum 上。
  • means[i] = sum / ncols: 计算平均值，并将其存储到 means 数组中。
  • return means: 返回包含每一行平均值的 NumPy 数组。
• setup.py:
  • from setuptools import setup: 导入 setuptools 模块，用于构建和安装 Python 包。
  • from Cython.Build import cythonize: 导入 cythonize 函数，用于将 Cython 代码编译成 C 代码。
  • import numpy: 导入 NumPy 库。
  • setup(...): 调用 setup 函数来配置构建过程。
    • ext_modules = cythonize("row_mean.pyx"): 指定要编译的 Cython 文件。
    • include_dirs=[numpy.get_include()]: 指定 NumPy 的头文件目录，以便在编译过程中找到 NumPy 的 C API。

性能比较：

用 %timeit 魔法命令分别测试 Python 和 Cython 实现的性能：

import pandas as pd
import numpy as np
import row_mean
import timeit

# 创建一个示例 DataFrame (增大规模)
data = np.random.rand(10000, 500)  # 10000 行，500 列
df = pd.DataFrame(data)

# 测试 Python 实现的性能
time_python = timeit.timeit(lambda: df.mean(axis=1), number=10)
print(f"Python (Pandas) mean time: {time_python/10:.4f} seconds")

# 测试 Cython 实现的性能
time_cython = timeit.timeit(lambda: row_mean.calculate_row_mean_cython(df), number=10)
print(f"Cython mean time: {time_cython/10:.4f} seconds")

你会发现，Cython 实现的速度明显快于 Python 实现。

加速 NumPy：数组操作的福音

NumPy 是 Python 中用于科学计算的核心库，提供了强大的数组操作功能。Cython 同样可以用来加速 NumPy 的数组操作。

场景：计算两个 NumPy 数组的点积

点积是线性代数中的基本操作，在机器学习和数据分析中经常用到。

1. Python (NumPy) 实现：

import numpy as np

def dot_product_python(a, b):
    """
    使用 NumPy 计算两个数组的点积。
    """
    return np.dot(a, b)

# 创建两个示例数组
a = np.random.rand(1000)
b = np.random.rand(1000)

# 测试 Python 实现
result_python = dot_product_python(a, b)
print(result_python)

2. Cython 实现：

• 创建一个名为 dot_product.pyx 的文件，内容如下：

# cython: language_level=3
import numpy as np
cimport numpy as cnp

def dot_product_cython(cnp.ndarray[cnp.float64_t, ndim=1] a, cnp.ndarray[cnp.float64_t, ndim=1] b):
    """
    使用 Cython 计算两个数组的点积。
    """
    cdef int n = len(a)
    cdef double result = 0.0
    cdef int i

    for i in range(n):
        result += a[i] * b[i]

    return result

• 修改 setup.py 文件，编译 Cython 代码：

from setuptools import setup
from Cython.Build import cythonize
import numpy

setup(
    ext_modules = cythonize("dot_product.pyx"),
    include_dirs=[numpy.get_include()]
)

• 编译 Cython 代码：

python setup.py build_ext --inplace

• 在 Python 中调用 Cython 代码：

import numpy as np
import dot_product

def dot_product_python(a, b):
    """
    使用 NumPy 计算两个数组的点积。
    """
    return np.dot(a, b)

# 创建两个示例数组
a = np.random.rand(1000)
b = np.random.rand(1000)

# 测试 Python 实现
result_python = dot_product_python(a, b)

# 测试 Cython 实现
result_cython = dot_product.dot_product_cython(a, b)

# 验证结果是否一致
np.testing.assert_allclose(result_python, result_cython)

print("结果一致")

代码解释：

• dot_product.pyx:
  • def dot_product_cython(cnp.ndarray[cnp.float64_t, ndim=1] a, cnp.ndarray[cnp.float64_t, ndim=1] b):: 定义一个函数，接受两个 NumPy 数组作为输入，并指定它们的类型为 float64，维度为 1。这里直接声明输入参数的类型，Cython编译器可以利用这些信息生成更高效的C代码。
  • cdef int n = len(a): 获取数组的长度，并将其声明为 C 类型的整数。
  • cdef double result = 0.0: 声明一个 C 类型的双精度浮点数，用于存储点积的结果。
  • for i in range(n):: 使用 C 风格的 for 循环遍历数组。
  • result += a[i] * b[i]: 计算点积。

性能比较：

用 %timeit 魔法命令分别测试 Python 和 Cython 实现的性能：

import numpy as np
import dot_product
import timeit

# 创建两个示例数组 (增大规模)
a = np.random.rand(1000000)
b = np.random.rand(1000000)

# 测试 Python 实现的性能
time_python = timeit.timeit(lambda: np.dot(a, b), number=10)
print(f"Python (NumPy) dot product time: {time_python/10:.4f} seconds")

# 测试 Cython 实现的性能
time_cython = timeit.timeit(lambda: dot_product.dot_product_cython(a, b), number=10)
print(f"Cython dot product time: {time_cython/10:.4f} seconds")

你会发现，Cython 实现的速度同样明显快于 Python 实现。

Cython 优化技巧：让你的代码飞起来

• 类型声明： 尽可能地使用 cdef 关键字声明变量的类型，让 Cython 编译器生成更高效的 C 代码。
• 避免 Python 对象： 尽量避免在 Cython 代码中使用 Python 对象，因为 Python 对象的访问速度较慢。
• 使用 NumPy 的 C API： 直接使用 NumPy 的 C API 可以避免 Python 对象的开销，提高性能。
• 减少函数调用： 函数调用会带来额外的开销，尽量减少函数调用的次数。
• 循环优化： 尽量使用 C 风格的 for 循环，避免使用 Python 的 for 循环。
• 内存管理： 注意内存管理，避免内存泄漏。

注意事项：

• 代码调试： Cython 代码的调试相对困难，可以使用 gdb 或 pdb 等调试工具。
• 编译问题： 编译 Cython 代码时可能会遇到各种问题，需要仔细阅读错误信息，并根据提示进行修改。
• 代码维护： Cython 代码的维护成本较高，需要谨慎使用。

总结：Cython，你值得拥有！

Cython 是一个强大的工具，可以用来加速 Pandas 和 NumPy 的操作，提高数据分析的效率。虽然 Cython 的学习曲线相对陡峭，但掌握它之后，你就可以轻松地编写高性能的 C 扩展，让你的 Python 代码飞起来！ 当然，也要记住，并非所有代码都需要 Cython 加速，只有性能瓶颈部分才需要考虑。过度优化反而会降低代码的可读性和可维护性。

最后的叮嘱：

实践是检验真理的唯一标准。多写代码，多尝试，你才能真正掌握 Cython 的精髓。

希望今天的讲座对大家有所帮助！ 下课！