python 性能的优化

一、使用jit让python的速度快100倍

NumPy的创始人Travis,创建了CONTINUUM,致力于将Python大数据处理方面的应用。
推出的Numba项目能够将处理NumPy数组的Python函数JIT编译为==机器码执行==,从而上百倍的提高程序的运算速度。

import time
from numba import jit

@jit
def foo(x,y):
    tt = time.time()
    s = 0
    for i in range(x,y):
        s += i
    print('Time used: {} sec'.format(time.time()-tt))
    return s

> print(foo(1,100000000))

numba中提供了一些修饰器,它们可以将其修饰的函数JIT编译成机器码函数,并返回一个可在Python中调用机器码的包装对象。为了能将Python函数编译成能高速执行的机器码,我们需要告诉JIT编译器函数的各个参数和返回值的类型。我们可以通过多种方式指定类型信息,在上面的例子中,类型信息由一个字符串’f8(f8[:])’指定。其中’f8’表示8个字节双精度浮点数,括号前面的’f8’表示返回值类型,括号里的表示参数类型,’[:]’表示一维数组。

import time
import numba as nb
from numba import jit

@jit('f8(f8[:])')
def sum1d(array):
    s = 0.0
    n = array.shape[0]
    for i in range(n):
        s += array[i]
    return s

import numpy as np
tt = time.time()
array = np.random.random(100000000)
sum1d(array)
np.sum(array)
sum(array)
print('Time used: {} sec'.format(time.time()-tt))

JIT能针对所有类型的参数进行运算,可以使用autojit:

from numba import autojit
@autojit
def sum1d2(array):
s = 0.0
n = array.shape[0]
for i in range(n):
s += array[i]
return s

%timeit sum1d2(array)
print sum1d2(np.ones(10, dtype=np.int32))
print sum1d2(np.ones(10, dtype=np.float32))
print sum1d2(np.ones(10, dtype=np.float64)

二、NumPy、Numba和Python异步编程的高性能大数据分析与对比

(1)NumPy是用于科学计算的基础Python包。它提供了强大的N维数组对象和复杂的(广播)功能。”导入NumPy库之后,Python程序的性能更好、执行速度更快、更容易保证一致性并能方便地使用大量的数学运算和矩阵功能。也许正因为如此,我们不再需要使用Python List对象了?重要的是,许多Python数据生态系统库都基于NumPy之上,像Pandas、SciPy、Matplotlib等等。
(2)asyncio应用场景应该是高IO负载下减少多线程切换的代价,由于PYTHON的特点,不管是用多线程,还是asyncio,都无法提高纯计算类任务的执行效率。
(3)Numba提供了由Python直接编写的高性能函数来加速应用程序的能力。通过几个注释,面向数组和数学计算较多的Python代码就可以被实时编译为原生机器指令。而且Numba拥有类似于C、C++和FORTRAN的性能,无需切换语言或Python解释器。

asyncio

asyncio是Python3.4版本引入的标准库,直接内置了对异步IO的支持。
asyncio的编程模型就是一个消息循环。我们从asyncio模块中直接获取一个EventLoop的引用,然后把需要执行的协程扔到EventLoop中执行,就实现了异步IO。

import asyncio

@asyncio.coroutine
def hello():
    print("Hello world!")
    # 异步调用asyncio.sleep(1):
    r = yield from asyncio.sleep(1)
    print("Hello again!")

# 获取EventLoop:
loop = asyncio.get_event_loop()
# 执行coroutine
loop.run_until_complete(hello())
loop.close()

@asyncio.coroutine把一个generator标记为coroutine类型,然后,我们就把这个coroutine扔到EventLoop中执行。

hello()会首先打印出Hello world!,然后,yield from语法可以让我们方便地调用另一个generator。由于asyncio.sleep()也是一个coroutine,所以线程不会等待asyncio.sleep(),而是直接中断并执行下一个消息循环。当asyncio.sleep()返回时,线程就可以从yield from拿到返回值(此处是None),然后接着执行下一行语句。

把asyncio.sleep(1)看成是一个耗时1秒的IO操作,在此期间,主线程并未等待,而是去执行EventLoop中其他可以执行的coroutine了,因此可以实现并发执行。

import threading
import asyncio

@asyncio.coroutine
def hello():
    print('Hello world! (%s)' % threading.currentThread())
    yield from asyncio.sleep(1)
    print('Hello again! (%s)' % threading.currentThread())

loop = asyncio.get_event_loop()
tasks = [hello(), hello()]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()

然后在coroutine内部用yield from调用另一个coroutine实现异步操作。也可以使用async和await来异步操作

async def hello():
    print("Hello world!")
    r = await asyncio.sleep(1)
    print("Hello again!")

分布式进程

在Thread和Process中,应当优选Process,因为Process更稳定,而且,Process可以分布到多台机器上,而Thread最多只能分布到同一台机器的多个CPU上。

Python的multiprocessing模块不但支持多进程,其中managers子模块还支持把多进程分布到多台机器上。一个服务进程可以作为调度者,将任务分布到其他多个进程中,依靠网络通信。由于managers模块封装很好,不必了解网络通信的细节,就可以很容易地编写分布式多进程程序。

举个例子:如果我们已经有一个通过Queue通信的多进程程序在同一台机器上运行,现在,由于处理任务的进程任务繁重,希望把发送任务的进程和处理任务的进程分布到两台机器上。怎么用分布式进程实现?

原有的Queue可以继续使用,但是,通过managers模块把Queue通过网络暴露出去,就可以让其他机器的进程访问Queue了。

我们先看服务进程,服务进程负责启动Queue,把Queue注册到网络上,然后往Queue里面写入任务:

# task_master.py

import random, time, queue
from multiprocessing.managers import BaseManager

# 发送任务的队列:
task_queue = queue.Queue()
# 接收结果的队列:
result_queue = queue.Queue()

# 从BaseManager继承的QueueManager:
class QueueManager(BaseManager):
    pass

# 把两个Queue都注册到网络上, callable参数关联了Queue对象:
QueueManager.register('get_task_queue', callable=lambda: task_queue)
QueueManager.register('get_result_queue', callable=lambda: result_queue)
# 绑定端口5000, 设置验证码'abc':
manager = QueueManager(address=('', 5000), authkey=b'abc')
# 启动Queue:
manager.start()
# 获得通过网络访问的Queue对象:
task = manager.get_task_queue()
result = manager.get_result_queue()
# 放几个任务进去:
for i in range(10):
    n = random.randint(0, 10000)
    print('Put task %d...' % n)
    task.put(n)
# 从result队列读取结果:
print('Try get results...')
for i in range(10):
    r = result.get(timeout=10)
    print('Result: %s' % r)
# 关闭:
manager.shutdown()
print('master exit.')

Master/Worker模型有什么用?其实这就是一个简单但真正的分布式计算,把代码稍加改造,启动多个worker,就可以把任务分布到几台甚至几十台机器上,比如把计算n*n的代码换成发送邮件,就实现了邮件队列的异步发送。

# task_worker.py

import time, sys, queue
from multiprocessing.managers import BaseManager

# 创建类似的QueueManager:
class QueueManager(BaseManager):
    pass

# 由于这个QueueManager只从网络上获取Queue,所以注册时只提供名字:
QueueManager.register('get_task_queue')
QueueManager.register('get_result_queue')

# 连接到服务器,也就是运行task_master.py的机器:
server_addr = '127.0.0.1'
print('Connect to server %s...' % server_addr)
# 端口和验证码注意保持与task_master.py设置的完全一致:
m = QueueManager(address=(server_addr, 5000), authkey=b'abc')
# 从网络连接:
m.connect()
# 获取Queue的对象:
task = m.get_task_queue()
result = m.get_result_queue()
# 从task队列取任务,并把结果写入result队列:
for i in range(10):
    try:
        n = task.get(timeout=1)
        print('run task %d * %d...' % (n, n))
        r = '%d * %d = %d' % (n, n, n*n)
        time.sleep(1)
        result.put(r)
    except Queue.Empty:
        print('task queue is empty.')
# 处理结束:
print('worker exit.')

Python的分布式进程接口简单,封装良好,适合需要把繁重任务分布到多台机器的环境下。

注意Queue的作用是用来传递任务和接收结果,每个任务的描述数据量要尽量小。比如发送一个处理日志文件的任务,就不要发送几百兆的日志文件本身,而是发送日志文件存放的完整路径,由Worker进程再去共享的磁盘上读取文件。

    原文作者:张晓天a
    原文地址: https://www.jianshu.com/p/da0f914ad26f
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞