Pandarallel 一个能让你的Pandas计算火力拉满的工具 - 大数据学习,java,hadoop,spark,hive,数据仓库

没有使用 Pandarallel
使用了 Pandarallel

众所周知，由于GIL的存在，Python单进程中的所有操作都是在一个CPU核上进行的，所以为了提高运行速度，我们一般会采用多进程的方式。而多进程无非就是以下几种方案：

multiprocessing
concurrent.futures.ProcessPoolExecutor()
joblib
ppserver
celery

这些方案对于普通Pandas玩家来说都不是特别友好，怎样才能算作一个友好的并行处理方案？

那就是原来的逻辑我基本不用变，仅修改需要计算的那行就能完成我们目标的方案，而 pandarallel 就是一个这样友好的工具。

没有并行计算（原始pandas）

pandarallel

df.apply(func)

df.parallel_apply(func)

df.applymap(func)

df.parallel_applymap(func)

df.groupby(args).apply(func)

df.groupby(args).parallel_apply(func)

df.groupby(args1).col_name.rolling(args2).apply(func)

df.groupby(args1).col_name.rolling(args2).parallel_apply(func)

df.groupby(args1).col_name.expanding(args2).apply(func)

df.groupby(args1).col_name.expanding(args2).parallel_apply(func)

series.map(func)

series.parallel_map(func)

series.apply(func)

series.parallel_apply(func)

series.rolling(args).apply(func)

series.rolling(args).parallel_apply(func)

可以看到，在 pandarallel 的世界里，你只需要替换原有的 pandas 处理语句就能实现多CPU并行计算。非常方便、非常nice.

在4核CPU的性能测试上，它比原始语句快了接近4倍。测试条件（OS: Linux Ubuntu 16.04，Hardware: Intel Core i7 @ 3.40 GHz – 4 cores），这就是我所说的，它把CPU充分利用了起来。

下面就给大家介绍这个模块怎么用，其实非常简单，任何代码只需要加几行代码就能实现质的飞跃。

1.准备

首先需要你的电脑安装好了Python环境，并且安装好了Python开发工具。

如果你还没有安装，可以参考以下文章：

如果仅用Python来处理数据、爬虫、数据分析或者自动化脚本、机器学习等，建议使用Python基础环境+jupyter即可，安装使用参考Windows/Mac 安装、使用Python环境+jupyter notebook

如果想利用Python进行web项目开发等，建议使用Python基础环境+Pycharm，安装使用参考：Windows下安装、使用Pycharm教程，这下全了和 Mac下玩转Python-安装&使用Python/PyCharm 。

请选择以下任一种方式输入命令安装依赖：
1. Windows 环境打开 Cmd (开始-运行-CMD)。
2. MacOS 环境打开 Terminal (command+空格输入Terminal)。
3. 如果你用的是 VSCode编辑器或 Pycharm，可以直接使用界面下方的Terminal.

pip install pandarallel

对于windows用户，有一个不好的消息是，它只能在Windows的linux子系统上运行（WSL），你可以在微软官网上找到安装教程：

https://docs.microsoft.com/zh-cn/windows/wsl/about

2.使用Pandarallel

使用前，需要对Pandarallel进行初始化：

from pandarallel import pandarallel
pandarallel.initialize()

这样才能调用并行计算的API，不过 initialize 中有一个重要参数需要说明，那就是 nb_workers ，它将指定并行计算的Worker数，如果没有设置，所有CPU的核都会用上。

Pandarallel一共支持8种Pandas操作，下面是一个apply方法的例子。

import pandas as pd
import time
import math
import numpy as np
from pandarallel import pandarallel

# 初始化
pandarallel.initialize()
df_size = int(5e6)
df = pd.DataFrame(dict(a=np.random.randint(1, 8, df_size),
                       b=np.random.rand(df_size)))
def func(x):
    return math.sin(x.a**2) + math.sin(x.b**2)

# 正常处理
res = df.apply(func, axis=1)

# 并行处理
res_parallel = df.parallel_apply(func, axis=1)

# 查看结果是否相同
res.equals(res_parallel)

其他方法使用上也是类似的，在原始的函数名称前加上 parallel_，比如 DataFrame.groupby.apply：

import pandas as pd
import time
import math
import numpy as np
from pandarallel import pandarallel

# 初始化
pandarallel.initialize()
df_size = int(3e7)
df = pd.DataFrame(dict(a=np.random.randint(1, 1000, df_size),
                       b=np.random.rand(df_size)))
def func(df):
    dum = 0
    for item in df.b:
        dum += math.log10(math.sqrt(math.exp(item**2)))
        
    return dum / len(df.b)

# 正常处理
res = df.groupby("a").apply(func)
# 并行处理
res_parallel = df.groupby("a").parallel_apply(func)
res.equals(res_parallel)

又比如 DataFrame.groupby.rolling.apply：

import pandas as pd
import time
import math
import numpy as np
from pandarallel import pandarallel

# 初始化
pandarallel.initialize()
df_size = int(1e6)
df = pd.DataFrame(dict(a=np.random.randint(1, 300, df_size),
                       b=np.random.rand(df_size)))
def func(x):
    return x.iloc[0] + x.iloc[1] ** 2 + x.iloc[2] ** 3 + x.iloc[3] ** 4

# 正常处理
res = df.groupby('a').b.rolling(4).apply(func, raw=False)
# 并行处理
res_parallel = df.groupby('a').b.rolling(4).parallel_apply(func, raw=False)
res.equals(res_parallel)

案例都是类似的，这里就直接列出表格，不浪费大家宝贵的时间去阅读一些重复的例子了: