如何加速pandas的DataFrame

 使用pandarallel模塊對 Pandas加速

python的dataFrame確實好用，但是明顯只能單核運算

使用pandas，當您運行以下行時：

# Standard apply

df.apply(func)

得到這個CPU使用率：

 

 

即使計算機有多個CPU，也只有一個完全專用於計算。

最近受群友推薦開始找到這個加速器，真的牛叉！！！可以真正體驗到用python也可單機核力全開，八核起飛的快感！！

【Pandaral·lel 】的想法是將pandas計算分布在計算機上所有可用的CPU上，以顯着提高速度。

安裝：

$ pip install pandarallel [--user]

導入和初始化：

導入：

from pandarallel import pandarallel

Initialization

pandarallel.initialize()

用法十分簡單：

用法：

使用帶有pandas DataFrame的簡單用例df和要應用的函數func，只需替換經典apply的parallel_apply。

# Standard pandas apply

df.apply(func)

# Parallel apply

df.parallel_apply(func)

 

• shm_size_mb: Deprecated.
• nb_workers: Number of workers used for parallelization. (int) If not set, all available CPUs will be used.
• progress_bar: Display progress bars if set to True. (bool, False by default)
• verbose: The verbosity level (int, 2 by default)
  • 0 - don't display any logs
  • 1 - display only warning logs
  • 2 - display all logs
• use_memory_fs: (bool, None by default)
  • If set to None and if memory file system is available, Pandarallel will use it to transfer data between the main process and workers. If memory file system is not available, Pandarallel will default on multiprocessing data transfer (pipe).
  • If set to True, Pandarallel will use memory file system to transfer data between the main process and workers and will raise a SystemError if memory file system is not available.
  • If set to False, Pandarallel will use multiprocessing data transfer (pipe) to transfer data between the main process and workers

With df a pandas DataFrame, series a pandas Series, func a function to apply/map, args, args1, args2 some arguments, and col_name a column name:

Without parallelization	With parallelization
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)
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)

You will find a complete example here for each row in this table.

報錯修復：pandarallel cannot find context for ‘fork‘

 進入到pandarallel 的包文件里，就是pandarallel 安裝的位置就是xxx:xxx/site-packages\pandarallel\pandarallel.py

 把fork改成spawn就行

# Python 3.8 on MacOS by default uses "spawn" instead of "fork" as start method for new
# processes, which is incompatible with pandarallel. We force it to use "fork" method.
context = get_context("spawn")

 

 

使用 joblib 模塊對 Pandas加速

如果需要對一個很大的數據集進行操作，而基於一列數據生成新的一列數據可能都需要耗費很長時間。

於是可以使用 joblib 進行並行處理。

假設我們有一個 dataframe 變量 data，要基於它的 source 列生成新的一列 double，其實就是把原來的 source 列做了個平方運算。

pandas中原生apply實現

import pandas as pd
from sqlalchemy import create_engine

mysql_engine=create_engine("""mysql+pymysql://root:xxxx@127.0.0.0.1:3306/tmp""")

df=pd.read_sql("ipesa_apply_a_loan_new",mysql_engine)

def double_func(xcreditScore):
    return pow(xcreditScore,2)

df["xcreditScore"] = df["xcreditScore"].apply(double_func)
df[['cust_id','xcreditScore']][1:10]

 

 

加速

from joblib import Parallel, delayed
import pandas as pd
from sqlalchemy import create_engine

mysql_engine=create_engine("""mysql+pymysql://root:xxxx@127.0.0.0.1:3306/tmp""")

df=pd.read_sql("ipesa_apply_a_loan_new",mysql_engine)

def double_func(xcreditScore):
    return pow(xcreditScore,2)

def key_func(subset):
    subset["xcreditScore"] = subset["xcreditScore"].apply(double_func)
    return subset


data_grouped = df.groupby(df.index)

results = Parallel(n_jobs=8)(delayed(key_func)(group) for index,group in data_grouped)

pd.concat(results)[['cust_id','xcreditScore']][1:10]

1.基本原理就是把整個 dataframe 根據 index，每行生成了一個子數據集。--使用df.groupby()函數使迭代返回值維持dataFrame的格式。

2.而把每個子數據集作為子任務使用多進程運行，最終生成 results 是多進程運行生成的結果的 list，使用 concat 重新把單個df組合起來。

  但是結果其實是耗時更長。我感覺很大原因是groupby聚合后迭代以及concat過程嚴重耗時。