Python代码性能爆发:四法优化极速进阶

文章标题:

Python代码性能跃升:四招优化极速进阶

文章内容:

编程领域中,性能优化是每位开发者都需掌握的重要技能。你是否曾思索过,除了更换算法,还有哪些高效策略能助力代码提速?本篇文章堪称典范,它通过四组优化组合,成功将一个普通函数的性能提升了330倍!作者不仅呈现了具体的优化方法,还分享了优化的理念。让我们跟随作者的思路,开启一场畅快的代码优化之旅。

作者:Itamar Turner-Trauring

译者:豌豆花下猫@Python猫

英文:330× faster: Four different ways to speed up your code

声明:本文翻译旨在交流学习,为便于阅读,部分内容稍作调整。转载请保留作者信息。

温馨提示: 本文原始版本与当前有所差异,比如曾提及500倍加速;本文已依据实际情况重新梳理,使论证更为清晰。

当你的Python代码运行迟缓如蜗牛,而你期望它能快如闪电时,实则存在诸多提速途径,从并行化到编译扩展应有尽有。若仅局限于一种方法,往往会错失提升良机,最终的代码也难以达到极致性能。

为不错过任何潜在的提速契机,我们可从“实践”维度来考量。每种实践:
- 以独特方式加速代码
- 涉及不同的技能与知识
- 可单独应用
- 也能组合运用,实现更大幅度的性能提升

为使这一点更为具象,本文将通过一个案例演示多种实践的应用,具体涵盖:
1. 效率(Efficiency):消除冗余或重复的计算。
2. 编译(Compilation):借助编译型语言,并巧妙规避编译器限制。
3. 并行化(Parallelism):充分发挥多核CPU的效能。
4. 流程(Process):采用能产出更快代码的开发流程。

我们将看到:
- 仅运用效率实践,就能实现近2倍的提速。
- 仅依靠编译实践,可达成10倍的提速。
- 两者结合,速度更上一层楼。
- 最后融入并行化实践,最终实现330倍的惊人加速。

我们的示例:统计字母频率

我们有一本英文书,简·奥斯汀的《诺桑觉寺》:

with open("northanger_abbey.txt") as f:
    TEXT = f.read()

我们的目标是分析书中字母的相对频率。元音是否比辅音更常见?哪个元音出现频率最高?

以下是最初的实现:

from collections import defaultdict

def frequency_1(text):
    # 一个键不存在时默认值为0的字典
    counts = defaultdict(lambda: 0)
    for character in text:
        if character.isalpha():
            counts[character.lower()] += 1
    return counts

运行结果如下:

sorted(
    (count, letter) for (letter, count)
    in frequency_1(TEXT).items()
)

[(1, 'à'),
 (2, 'é'),
 (3, 'ê'),
 (111, 'z'),
 (419, 'q'),
 (471, 'j'),
 (561, 'x'),
 (2016, 'k'),
 (3530, 'v'),
 (5297, 'b'),
 (5404, 'p'),
 (6606, 'g'),
 (7639, 'w'),
 (7746, 'f'),
 (7806, 'y'),
 (8106, 'c'),
 (8628, 'm'),
 (9690, 'u'),
 (13431, 'l'),
 (14164, 'd'),
 (20675, 's'),
 (21107, 'r'),
 (21474, 'h'),
 (22862, 'i'),
 (24670, 'n'),
 (26385, 'a'),
 (26412, 'o'),
 (30003, 't'),
 (44251, 'e')]

不出所料,出现频率最高的字母是 "e"。

那如何让这个函数运行得更快呢?

流程实践:测量与测试

软件开发不仅依赖源代码、库、解释器、编译器等“产物”,还离不开你的工作“流程”——即你做事的方法。性能优化亦是如此。本文将介绍优化过程中不可或缺的两种流程实践:
1. 通过基准测试和性能分析来测量代码运行速度。
2. 测试优化后的代码,确保其行为与原始版本一致。

我们可以先用 line_profiler 工具分析函数,找出最耗时的代码行:

Line #      Hits   % Time  Line Contents
========================================
     3                     def frequency_1(text):
     4                         # 一个键不存在时默认值为0的字典
     5                         # available:
     6         1      0.0      counts = defaultdict(lambda: 0)
     7    433070     30.4      for character in text:
     8    433069     27.3          if character.isalpha():
     9    339470     42.2              counts[character.lower()] += 1
    10         1      0.0      return counts

效率实践:减少无用功

效率实践的核心在于用更少的工作量达成相同结果。这类优化通常在较高抽象层面进行,无需关注底层CPU细节,适用于大多数编程语言。其本质是通过改变计算逻辑来减少浪费。

减少内循环的工作量

从上述性能分析可见,函数大部分时间消耗在 counts[character.lower()] += 1 这一行。显然,对每个字母都调用 character.lower() 是一种浪费。我们反复将 "I" 转为 "i",甚至将 "i" 转为 "i"。

优化思路:我们可先分别统计大写和小写字母的数量,最后再合并,而非每次都进行小写转换。

def frequency_2(text):
    split_counts = defaultdict(lambda: 0)
    for character in text:
        if character.isalpha():
            split_counts[character] += 1

    counts = defaultdict(lambda: 0)
    for character, num in split_counts.items():
        counts[character.lower()] += num
    return counts

# 确保新函数结果与旧函数完全一致
assert frequency_1(TEXT) == frequency_2(TEXT)

说明:此处的 assert 属于流程实践的一部分。一个更快但结果错误的函数毫无意义。尽管最终文章中看不到这些断言,但它们在开发时帮我们排查出了诸多bug。

基准测试(流程实践的一环)显示,此优化确实使代码运行更快:

| frequency_1(TEXT) | 34,592.5 µs |
| frequency_2(TEXT) | 25,798.6 µs |

针对特定数据和目标进行优化

我们继续运用效率实践,此次针对具体目标和数据进一步优化。查看最新代码的性能分析:

Line #      Hits   % Time  Line Contents
========================================
     3                     def frequency_2(text):
     4         1      0.0      split_counts = defaultdict(lambda: 0)
     5    433070     33.6      for character in text:
     6    433069     32.7          if character.isalpha():
     7    339470     33.7              split_counts[character] += 1
     8
     9         1      0.0      counts = defaultdict(lambda: 0)
    10        53      0.0      for character, num in split_counts.items():
    11        52      0.0          counts[character.lower()] += num
    12         1      0.0      return counts

可见,split_counts[character] += 1 仍是耗时大户。如何加速?答案是用 list 替换 defaultdict(本质为 dict)。list 的索引速度远快于 dict
- list 存储条目仅需一次数组索引
- dict 需计算哈希、可能多次比较,还需内部数组索引

list 的索引必须为整数,不能像 dict 那样用字符串,所以需将字符转为数字。幸运的是,每个字符都可通过 ord() 查得数值:

ord('a'), ord('z'), ord('A'), ord('Z')
# (97, 122, 65, 90)

chr() 可将数值转回字符:

chr(97), chr(122)
# ('a', 'z')

因此可用 my_list[ord(character)] += 1 计数。但前提是提前知晓 list 的大小。若处理任意字母字符,list 可能较大:

ideograph = '𫞉'
ord(ideograph), ideograph.isalpha()
# (178057, True)

再回顾我们的目标:
1. 处理对象是英文文本,这是题目要求。
2. 输出结果中确实有少量非标准英文字母(如 'à'),但极为罕见。(严格来说 'à' 应归为 'a',但此处未做处理……)
3. 我们仅关心相对频率,而非绝对精确计数。

基于此,我们决定简化问题:仅统计 'A' 到 'Z',忽略其他字符,包括带重音的。对于英文文本而言,这几乎不影响字母相对频率。

如此问题便简化了:字符集有限且已知,可放心用 list 替代 dict

优化后实现如下:

def frequency_3(text):
    # 创建长度为128的零列表;ord('z')是122,128足够
    split_counts = [0] * 128
    for character in text:
        index = ord(character)
        if index < 128:
            split_counts[index] += 1

    counts = {}
    for letter in 'abcdefghijklmnopqrstuvwxyz':
        counts[letter] = (
            split_counts[ord(letter)] +
            split_counts[ord(letter.upper())]
        )
    return counts

由于输出仅包含A到Z,正确性检查需稍作调整:

def assert_matches(counts1, counts2):
    """确保A到Z的计数匹配"""
    for character in 'abcdefghijklmnopqrstuvwxyz':
        assert counts1[character] == counts2[character]

assert_matches(
    frequency_1(TEXT),
    frequency_3(TEXT)
)

新实现运行更快:

| frequency_2(TEXT) | 25,965.5 µs |
| frequency_3(TEXT) | 19,443.5 µs |

编译实践:切换到更快的语言

接下来我们切换到编译型语言——Rust。

实际上可直接将 frequency_1() 移植到 Rust,编译器会自动完成Python中需手动优化的部分。

但大多数情况下,无论使用何种语言,效率实践都需自行完成。这也是“效率”与“编译”为两种不同实践的原因:它们带来的性能提升来源不同。我们在 frequency_2()frequency_3() 中所做的优化,同样能使Rust代码运行更快。

为证明这一点,我将上述三个Python函数均移植到了Rust(前两个源码可点击展开查看):🦄

前两个版本在Rust中的实现

#[pyfunction]
fn frequency_1_rust(
    text: &str,
) -> PyResult<HashMap<char, u32>> {
    let mut counts = HashMap::new();
    for character in text.chars() {
        if character.is_alphabetic() {
            *counts
                .entry(
                    character
                        .to_lowercase()
                        .next()
                        .unwrap_or(character),
                )
                .or_default() += 1;
        }
    }
    Ok(counts)
}

#[pyfunction]
fn frequency_2_rust(
    text: &str,
) -> PyResult<HashMap<char, u32>> {
    let mut split_counts: HashMap<char, u32> =
        HashMap::new();
    for character in text.chars() {
        if character.is_alphabetic() {
            *split_counts.entry(character).or_default() +=
                1;
        }
    }

    let mut counts = HashMap::new();
    for (character, num) in split_counts.drain() {
        *counts
            .entry(
                character
                    .to_lowercase()
                    .next()
                    .unwrap_or(character),
            )
            .or_default() += num;
    }
    Ok(counts)
}

第三个版本在Rust里的样子:

fn ascii_arr_to_letter_map(
    split_counts: [u32; 128],
) -> HashMap<char, u32> {
    let mut counts: HashMap<char, u32> = HashMap::new();
    for index in ('a' as usize)..=('z' as usize) {
        let character =
            char::from_u32(index as u32).unwrap();
        let upper_index =
            character.to_ascii_uppercase() as usize;
        counts.insert(
            character,
            split_counts[index] + split_counts[upper_index],
        );
    }
    counts
}

#[pyfunction]
fn frequency_3_rust(text: &str) -> HashMap<char, u32> {
    let mut split_counts = [0u32; 128];
    for character in text.chars() {
        let character = character as usize;
        if character < 128 {
            split_counts[character] += 1;
        }
    }

    ascii_arr_to_letter_map(split_counts)
}

所有三个Rust版本的结果均与Python版本一致:

assert_matches(frequency_1(TEXT), frequency_1_rust(TEXT))
assert_matches(frequency_1(TEXT), frequency_2_rust(TEXT))
assert_matches(frequency_1(TEXT), frequency_3_rust(TEXT))

对所有6个版本进行基准测试,清晰表明效率实践编译实践的性能优势是不同且互补的。能加速Python代码的效率优化,同样能加速Rust代码。

函数 运行时间 (µs)
frequency_1(TEXT) 33,741.5
frequency_2(TEXT) 25,797.4
frequency_3(TEXT) 19,432.0
frequency_1_rust(TEXT) 3,704.3
frequency_2_rust(TEXT) 3,504.8
frequency_3_rust(TEXT) 204.9

简言之:效率和编译是两种不同的速度来源。

并行化实践:榨干多核CPU

至此,代码均在单核CPU上运行。但如今的电脑大多具有多核,利用并行计算是另一种速度来源,因此它也是独立的实践。

以下是用 Rayon 库 实现的Rust并行版本:

```rust
fn sum(mut a: [u32; 128], b: [u32; 128]) -> [u32; 128] {
for i in 0..128 {
a[i] += b[i];
}
a
}

[pyfunction]

fn frequency_parallel_rust(
py: Python<'_>,
text: &str,
) -> HashMap {
use rayon::prelude::*;

// 确保释放全局解释器锁(GIL)
let split_counts = py.allow_threads(|| {
    // 一个榨取 Rayon 更多性能的技巧:
    // 我们关心的 ASCII 字符总是由单个字节明确表示。
    // 所以直接处理字节是安全的,这能让我们强制 Rayon 使用数据块。
    text.as_bytes()
        // 并行迭代数据块
        .par_chunks(8192)
        .fold_with(
            [0u32; 128],
            |mut split_counts, characters| {
                for character in characters {
                    if *character < 128 {
版权声明:程序员胖胖胖虎阿 发表于 2025年8月4日 下午7:42。
转载请注明:Python代码性能爆发:四法优化极速进阶 | 胖虎的工具箱-编程导航

相关文章

暂无评论

暂无评论...