詳解pandas apply 並行處理的幾種方法

1. pandarallel (pip install )

對於一個帶有Pandas DataFrame df的簡單用例和一個應用func的函數,隻需用parallel_apply替換經典的apply。

from pandarallel import pandarallel
 
# Initialization
pandarallel.initialize()
 
# Standard pandas apply
df.apply(func)
 
# Parallel apply
df.parallel_apply(func)

註意,如果不想並行化計算,仍然可以使用經典的apply方法。

另外可以通過在initialize函數中傳遞progress_bar=True來顯示每個工作CPU的一個進度條。

2. joblib (pip install )

 https://pypi.python.org/pypi/joblib

# Embarrassingly parallel helper: to make it easy to write readable parallel code and debug it quickly
 
from math import sqrt
from joblib import Parallel, delayed
 
def test():
  start = time.time()
  result1 = Parallel(n_jobs=1)(delayed(sqrt)(i**2) for i in range(10000))
  end = time.time()
  print(end-start)
  result2 = Parallel(n_jobs=8)(delayed(sqrt)(i**2) for i in range(10000))
  end2 = time.time()
  print(end2-end)

——-輸出結果———-

0.4434356689453125
0.6346755027770996

3. multiprocessing

import multiprocessing as mp
 
with mp.Pool(mp.cpu_count()) as pool:
  df['newcol'] = pool.map(f, df['col'])
multiprocessing.cpu_count()

返回系統的CPU數量。

該數量不同於當前進程可以使用的CPU數量。可用的CPU數量可以由 len(os.sched_getaffinity(0)) 方法獲得。

可能引發 NotImplementedError 。

參見os.cpu_count()

4. 幾種方法性能比較

(1)代碼

import sys
import time
import pandas as pd
import multiprocessing as mp
from joblib import Parallel, delayed
from pandarallel import pandarallel
from tqdm import tqdm, tqdm_notebook
 
 
def get_url_len(url):
  url_list = url.split(".")
  time.sleep(0.01) # 休眠0.01秒
  return len(url_list)
 
def test1(data):
  """
  不進行任何優化
  """
  start = time.time()
  data['len'] = data['url'].apply(get_url_len)
  end = time.time()
  cost_time = end - start
  res = sum(data['len'])
  print("res:{}, cost time:{}".format(res, cost_time))
 
def test_mp(data):
  """
  采用mp優化
  """
  start = time.time()
  with mp.Pool(mp.cpu_count()) as pool:
    data['len'] = pool.map(get_url_len, data['url'])
  end = time.time()
  cost_time = end - start
  res = sum(data['len'])
  print("test_mp \t res:{}, cost time:{}".format(res, cost_time))
 
def test_pandarallel(data):
  """
  采用pandarallel優化
  """
  start = time.time()
  pandarallel.initialize()
  data['len'] = data['url'].parallel_apply(get_url_len)
  end = time.time()
  cost_time = end - start
  res = sum(data['len'])
  print("test_pandarallel \t res:{}, cost time:{}".format(res, cost_time))
 
 
def test_delayed(data):
  """
  采用delayed優化
  """
  def key_func(subset):
    subset["len"] = subset["url"].apply(get_url_len)
    return subset
 
  start = time.time()
  data_grouped = data.groupby(data.index)
  # data_grouped 是一個可迭代的對象,那麼就可以使用 tqdm 來可視化進度條
  results = Parallel(n_jobs=8)(delayed(key_func)(group) for name, group in tqdm(data_grouped))
  data = pd.concat(results)
  end = time.time()
  cost_time = end - start
  res = sum(data['len'])
  print("test_delayed \t res:{}, cost time:{}".format(res, cost_time))
 
 
if __name__ == '__main__':
  
  columns = ['title', 'url', 'pub_old', 'pub_new']
  temp = pd.read_csv("./input.csv", names=columns, nrows=10000)
  data = temp
  """
  for i in range(99):
    data = data.append(temp)
  """
  print(len(data))
  """
  test1(data)
  test_mp(data)
  test_pandarallel(data)
  """
  test_delayed(data)

(2) 結果輸出

1k
res:4338, cost time:0.0018074512481689453
test_mp   res:4338, cost time:0.2626469135284424
test_pandarallel   res:4338, cost time:0.3467681407928467
 
1w
res:42936, cost time:0.008773326873779297
test_mp   res:42936, cost time:0.26111721992492676
test_pandarallel   res:42936, cost time:0.33237743377685547
 
10w
res:426742, cost time:0.07944369316101074
test_mp   res:426742, cost time:0.294996976852417
test_pandarallel   res:426742, cost time:0.39208269119262695
 
100w
res:4267420, cost time:0.8074917793273926
test_mp   res:4267420, cost time:0.9741342067718506
test_pandarallel   res:4267420, cost time:0.6779992580413818
 
1000w
res:42674200, cost time:8.027287006378174
test_mp   res:42674200, cost time:7.751036882400513
test_pandarallel   res:42674200, cost time:4.404983282089233

在get_url_len函數裡加個sleep語句(模擬復雜邏輯),數據量為1k,運行結果如下:

1k
res:4338, cost time:10.054503679275513
test_mp   res:4338, cost time:0.35697126388549805
test_pandarallel   res:4338, cost time:0.43415403366088867
test_delayed   res:4338, cost time:2.294757843017578

5. 小結

(1)如果數據量比較少,並行處理比單次執行效率更慢;

(2)如果apply的函數邏輯簡單,並行處理比單次執行效率更慢。

6. 問題及解決方法

(1)ImportError: This platform lacks a functioning sem_open implementation, therefore, the required synchronization primitives needed will not function, see issue 3770.

https://www.jianshu.com/p/0be1b4b27bde

(2)Linux查看物理CPU個數、核數、邏輯CPU個數

https://lover.blog.csdn.net/article/details/113951192

(3) 進度條的使用

https://www.jb51.net/article/206219.htm

到此這篇關於詳解pandas apply 並行處理的幾種方法的文章就介紹到這瞭,更多相關pandas apply 並行處理內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!

推薦閱讀: