赵走x博客 - 开发者的网上家园

Python性能分析与优化：7、性能分析示例

资源编号:75727 书籍 Python性能分析与优化热度：87

现在我们已经掌握了`cProfile`和`Stats`的基本用法了，下面来探索一些更加有趣且真实的例子吧。

### **2.2.4　性能分析示例**

现在我们已经掌握了`cProfile`和`Stats`的基本用法了，下面来探索一些更加有趣且真实的例子吧。

**1. 回到斐波那契数列**

让我们重新回到斐波那契数列，因为用递归方式计算的斐波那契数列有很大的改进空间。

让我们先看看未经性能分析也没有优化过的代码：

```
import profile

def fib(n):
    if n <= 1:
        return n
    else:
        return fib(n-1) + fib(n-2)

def fib_seq(n):
    seq = []
    if n > 0:
        seq.extend(fib_seq(n-1))
    seq.append(fib(n))
    return seq

profile.run('print fib_seq(20); print')

```

代码的输出结果如下：

![image.png](https://img.hacpai.com/file/2019/08/image-cba9787a.png)

虽然输出的结果打印没问题，但是看看上图中画框的部分。具体解释如下：

* 在0.142秒内，共有57 356个函数调用
    
* 一共只有66个原生调用（不包括递归）
    
* 在代码的第三行，一共有57 270（57 291-21）次递归函数调用

我们已经知道，过多的函数调用将增加额外的时间消耗。从图中可见（`cumtime`列），大部分时间都消耗在递归函数里了，因此我们有理由确信如果让递归函数加速，整个程序的执行时间也会改善。

现在，让我们给`fib`函数加一个简单的装饰器，缓存之前计算的值［这是一种函数返回值缓存（memoization）技术，将在后面的章节里介绍］，这样每个`fib`函数的值就不需要重复计算了：

```
import profile

class cached:
    def __init__(self, fn):
        self.fn = fn
        self.cache = {}

def __call__(self, *args):
        try:
            return self.cache[args]
        except KeyError:
            self.cache[args] = self.fn(*args)
            return self.cache[args]

@cached
def fib(n):
    if n <= 1:
        return n
    else:
        return fib(n-1) + fib(n-2)

def fib_seq(n):
    seq = []
    if n > 0:
        seq.extend(fib_seq(n-1))
    seq.append(fib(n))
    return seq

profile.run('print fib_seq(20); print')

```

现在，让我们运行代码，结果如下所示：

![image.png](https://img.hacpai.com/file/2019/08/image-a4361d2a.png)

程序的函数调用次数从57 000多下降到了145，运行时间也从0.142秒下降到了0.001秒。这是一个非常给力的优化！但是，我们的原生调用变多了，而递归调用明显减少了。

让我们再进行另一项优化。虽然我们的例子对单个函数调用的处理非常快，但是让我们试试把多个调用合成一组，用`stats`输出结果。这样做可能还会看到一些有趣的新结果。为此，我们需要使用`stats`模块。示例代码如下：

```
import cProfile
import pstats
from fibo4 import fib, fib_seq

filenames = []
profiler = cProfile.Profile()
profiler.enable()
for i in range(5):
    print fib_seq(1000); print
profiler.create_stats()
stats = pstats.Stats(profiler)
stats.strip_dirs().sort_stats('cumulative').print_stats()
stats.print_callers()

```

我们已经做了简化。计算1000个斐波那契数列可能计算量太大，尤其是递归实现时。这样运行代码肯定会超过`cPython`的递归限制。`cPython`为了防止栈溢出，设置了递归保护措施（理论上，这个问题可以通过尾递归解决，但是`cPython`没有提供）。因此，我们找到了另一个解决方案。让我们修复这个问题，运行下面的代码：

```
import profile

def fib(n):
    a, b = 0, 1
    for i in range(0, n):
        a, b = b, a+b
    return a

def fib_seq(n):
    seq = []
    for i in range(0, n + 1):
        seq.append(fib(i))
    return seq

print fib_seq(1000)

```

上面的代码会打印出存储很多数值的超长列表，但是这些行证明我们实现了目标。我们可以计算1000个斐波那契数列。现在，让我们分析看看结果如何。

用新的性能分析函数，不过需要迭代版的斐波那契实现，代码如下：

```
import cProfile
import pstats
from fibo_iter import fib, fib_seq

```

这段代码显示在命令行的结果如下：

![image.png](https://img.hacpai.com/file/2019/08/image-de5bb7ef.png)

新代码用0.187秒计算了1000个斐波那契数列5次。虽然这个结果并不差，但是我们知道可以用缓存数值来优化，和我们之前做的一样。你已经看到，`fib`函数调用了5005次，如果缓存结果，就可以大幅度降低调用次数，这意味着很少的运行时间。

只需要一点点努力，我们就可以通过缓存改善之前被调用了5005次的`fib`函数的调用时间：

```
import profile

class cached:
    def __init__(self, fn):
        self.fn = fn
        self.cache = {}

def __call__(self, *args):
        try:
            return self.cache[args]
        except KeyError:
            self.cache[args] = self.fn(*args)
            return self.cache[args]

@cached
def fib(n):
    a, b = 0, 1
    for i in range(0, n):
        a, b = b, a+b
    return a

def fib_seq(n):
    seq = []
    for i in range(0, n + 1):
        seq.append(fib(i))
    return seq

print fib_seq(1000)

```

你应该会得到如下的运行结果：

![image.png](https://img.hacpai.com/file/2019/08/image-a7ec91f0.png)

只要简单地缓存`fib`函数的结果，就可以把运行时间从0.187秒缩短到0.006秒。这是非常给力的优化！干得漂亮！

**2. 推文数据统计**

让我们再看一个内容上更复杂一些的例子，毕竟斐波那契数列不是现实中人人都会用到的。还是让我们再做一些有趣的事情吧。现在Twitter已经允许你以CSV格式下载自己的推文列表。我们就用自己的数据做一些统计。

通过获取的数据，我们可以统计下面的信息：

* 实际回复的信息占比
    
* 网站（[https://twitter.com](https://twitter.com)）发布的推文占比
    
* 手机发布的推文占比

我们的程序输出的结果应该如下图所示。

![image.png](https://img.hacpai.com/file/2019/08/image-c3fb74b9.png)

为了简便，我们重点关注CVS文件解析，并做一些基本计算。我们不用任何第三方模块，这样，我们就可以完全控制代码和分析的内容了。也就是说，像Python的`csv`模块我们也不用。

前面出现过的其他不太好的做法，比如`inc_stat`函数，或者在处理文件之前把整个文件都载入内存，都会提醒你，这只是一个显示基本改进方法的示例。

下面是初始代码：

```
def build_twit_stats():
    STATS_FILE = './files/tweets.csv'
    STATE = {
        'replies': 0,
        'from_web': 0,
        'from_phone': 0,
        'lines_parts': [],
        'total_tweets': 0
    }
    read_data(STATE, STATS_FILE)
    get_stats(STATE)
    print_results(STATE)

def get_percentage(n, total):
    return (n * 100) / total

def read_data(state, source):
    f = open(source, 'r')

lines = f.read().strip().split("\"\n\"")
    for line in lines:
       state['lines_parts'].append(line.strip().split(','))
    state['total_tweets'] = len(lines)

def inc_stat(state, st):
    state[st] += 1

def get_stats(state):
    for i in state['lines_parts']:
        if(i[1] != '""'):
            inc_stat(state, 'replies')
        if(i[4].find('Twitter Web Client') > -1):
            inc_stat(state, 'from_web')
        else:
            inc_stat(state, 'from_phone')

def print_results(state):
    print "-------- My twitter stats -------------"
    print "%s%% of tweets are replies" % (get_percentage(state['replies'], state ['total_tweets']))
    print "%s%% of tweets were made from the website" % (get_percentage(state ['from_web'], state['total_tweets']))
    print "%s%% of tweets were made from my phone" % (get_percentage(state['from_phone'], state['total_tweets']))

```

其实，这段代码并不是太复杂，就是读取文件内容，按行分割，再把每一行分配到不同的类型中，最后统计各个类型推文的数量。初看这段代码，可能会认为没什么可优化的，但我们会发现其实还是有优化空间的。

另一个需要注意的地方是，我们要处理的数据有150MB。

下面的代码会导入代码并使用它生成性能分析报告：

```
import cProfile
import pstats
from B02088_02_14 import build_twit_stats

profiler = cProfile.Profile()
profiler.enable()

build_twit_stats()

profiler.create_stats()
stats = pstats.Stats(profiler)
stats.strip_dirs().sort_stats('cumulative').print_stats()

```

执行代码获得的结果如下：

![image.png](https://img.hacpai.com/file/2019/08/image-92cc8f54.png)

上面的截屏中有三点需要注意：

(1) 程序的总执行时间

(2) 不同函数累计的调用次数

(3) 每个函数的总调用次数

我们的目标是减少总执行时间。因此，我们需要考虑不同函数调用的累计次数和每个函数的总调用次数。关于这两点，我们可以得出下面的结论。

* `build_twit_stats`消耗了最多的时间。然而，你会看到在前面的代码中，它只是调用了其他所有函数，这是显而易见的。我们可以把注意力集中在耗时第二多的`read_data`函数上。这倒是挺有意思，我们的性能瓶颈不是计算统计数据，而是读取文件。
    
* 在代码的第三行，我们可以清楚地看到`read_data`函数的瓶颈。我们使用了太多`split`命令，它们的时间累加了。
    
* 还可以看到第四耗时的函数`get_stats`。

那么现在让我们带着这些问题，看看有没有更好的解决方案。最大的性能瓶颈就是加载数据。我们首先把数据加载到内存中，然后重复地遍历文件计算统计数据。我们可以改成逐行读取文件，然后每读一行统计一次。让我们看看代码应该怎么写。

新的`read_data`函数应该像这样：

```
def read_data(state, source):
    f = open(source)

buffer_parts = []
    for line in f:
      # 由于多行推文在文件中也被保存为若干行，
      # 因此需要考虑把它们合并到一起。
      parts = line.split('","')
      buffer_parts += parts
      if len(parts) == 10:
        state['lines_parts'].append(buffer_parts)
        get_line_stats(state, buffer_parts)
        buffer_parts = []
    state['total_tweets'] = len(state['lines_parts'])

```

我们需要增加一些逻辑处理多行推文，也就是CSV文件中的多行记录。我们把`get_stats`函数改成了`get_line_stats`，这样做可以简化逻辑，因为它只计算当前行的值。

```
def get_line_stats(state, line_parts):
    if line_parts[1] != '' :
        state['replies'] += 1
    if 'Twitter Web Client' in line_parts[4]:
        state['from_web'] += 1
    else:
        state['from_phone'] += 1

```

最后两项改进，一是移除`inc_stat`函数的调用，由于我们用字典，所以调用就没必要了；二是利用`in`操作符替换`find`方法。

让我们再运行一次代码，看看变化：

![image.png](https://img.hacpai.com/file/2019/08/image-f35016f1.png)

运行时间从2秒降到了1.6秒，算是一个不错的改进。`read_data`函数仍然是耗时最多的函数，但是现在的原因是`get_line_stats`函数。我们还可以进一步优化它，虽然这个函数并没有做很多操作，但是在循环体中不断地调用它会消耗一些查询时间。我们可以对这个函数进行内联，看看效果有没有改善。

新的代码如下：

```
def read_data(state, source):
    f = open(source)

buffer_parts = []
    for line in f:
      # 由于多行推文在文件中也被保存为若干行，
      # 因此需要考虑把它们合并到一起。
      parts = line.split('","')
      buffer_parts += parts
      if len(parts) == 10:
        state['lines_parts'].append(buffer_parts)
        if buffer_parts[1] != '' :
          state['replies'] += 1
        if 'Twitter Web Client' in buffer_parts[4]:
          state['from_web'] += 1
        else:
          state['from_phone'] += 1
        buffer_parts = []
    state['total_tweets'] = len(state['lines_parts'])

```

现在，结果有了新变化，如下图所示。

![image.png](https://img.hacpai.com/file/2019/08/image-edfca5bc.png)

相比第一张图和前面那一张图，这是很明显的进步。我们把程序运行时间从2秒降到了1.4秒。函数调用次数也明显降低了（从大约300万次降到了170万次），相应地也减少了函数查询和调用的时间。

作为额外的改善，我们还可以简化代码以增加可读性。最终的代码如下：

def get_percentage(n, total):
    return (n * 100) / total

def read_data(state, source):
    f = open(source)

buffer_parts = []
    for line in f:
      # 由于多行推文在文件中也被保存为若干行，
      # 因此需要考虑把它们合并到一起。
      parts = line.split('","')
      buffer_parts += parts
      if len(parts) == 10:
        state['lines_parts'].append(buffer_parts)
        if buffer_parts[1] != '' :
          state['replies'] += 1
        if 'Twitter Web Client' in buffer_parts[4]:
          state['from_web'] += 1
        else:
          state['from_phone'] += 1
        buffer_parts = []
    state['total_tweets'] = len(state['lines_parts'])

def print_results(state):
    print "-------- My twitter stats -------------"
    print "%s%% of tweets are replies" % (get_percentage(state['replies'], state['total_tweets']))
    print "%s%% of tweets were made from the website" % (get_percentage(state ['from_web'], state['total_tweets']))
    print "%s%% of tweets were made from my phone" % (get_percentage(state ['from_phone'], state['total_tweets']))

```

下面我们对`cProfile`做个总结。通过它我们可以对代码进行性能分析，获取每个函数的调用次数和总调用次数。它帮助我们通过系统全局视角改进代码。下面将介绍另一种性能分析器，它可以为我们提供每一行代码的性能细节，这是`cProfile`无法提供的。