快捷導(dǎo)航

pandas apply多線程實(shí)現(xiàn)代碼

更新時(shí)間：2020年08月17日 09:38:08 作者：frostjsy

這篇文章主要介紹了pandas apply多線程實(shí)現(xiàn)代碼，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

一、多線程化選擇

并行化一個(gè)代碼有兩大選擇：multithread 和 multiprocess。

Multithread，多線程，同一個(gè)進(jìn)程（process）可以開啟多個(gè)線程執(zhí)行計(jì)算。每個(gè)線程代表了一個(gè) CPU 核心，這么多線程可以訪問同樣的內(nèi)存地址（所謂共享內(nèi)存），實(shí)現(xiàn)了線程之間的通訊，算是最簡(jiǎn)單的并行模型。

Multiprocess，多進(jìn)程，則相當(dāng)于同時(shí)開啟多個(gè) Python 解釋器，每個(gè)解釋器有自己獨(dú)有的數(shù)據(jù)，自然不會(huì)有數(shù)據(jù)沖突。

二、并行化思想

并行化的基本思路是把 dataframe 用 np.array_split 方法切割成多個(gè)子 dataframe。再調(diào)用 Pool.map 函數(shù)并行地執(zhí)行。注意到順序執(zhí)行的 pandas.DataFrame.apply 是如何轉(zhuǎn)化成 Pool.map 然后并行執(zhí)行的。

Pool 對(duì)象是一組并行的進(jìn)程，開源Pool類

開源Pool類定義

 def Pool(self, processes=None, initializer=None, initargs=(),
       maxtasksperchild=None):
    '''Returns a process pool object'''
    from .pool import Pool
    return Pool(processes, initializer, initargs, maxtasksperchild,
          context=self.get_context())

設(shè)置進(jìn)程初始化函數(shù)

def init_process(global_vars):
  global a
  a = global_vars

設(shè)置進(jìn)程初始化函數(shù)

Pool(processes=8,initializer=init_process,initargs=(a,))

其中，指定產(chǎn)生 8 個(gè)進(jìn)程，每個(gè)進(jìn)程的初始化需運(yùn)行 init_process函數(shù)，其參數(shù)為一個(gè) singleton tuple a. 利用 init_process 和 initargs，我們可以方便的設(shè)定需要在進(jìn)程間共享的全局變量（這里是 a）。

with 關(guān)鍵詞是 context manager，避免寫很繁瑣的處理開關(guān)進(jìn)程的邏輯。

 with Pool(processes=8,initializer=init_process,initargs=(a,)) as pool:    
    result_parts = pool.map(apply_f,df_parts)

三、多線程化應(yīng)用

多線程時(shí)間比較和多線程的幾種apply應(yīng)用

import numpy as np
import pandas as pd
import time
from multiprocessing import Pool

def f(row):
  #直接對(duì)某列進(jìn)行操作
  return sum(row)+a

def f1_1(row):
  #對(duì)某一列進(jìn)行操作，我這里的columns=range(0,2)，此處是對(duì)第0列進(jìn)行操作
  return row[0]**2

def f1_2(row1):
  #對(duì)某一列進(jìn)行操作，我這里的columns=range(0,2)，此處是對(duì)第0列進(jìn)行操作
  return row1**2

def f2_1(row):
  #對(duì)某兩列進(jìn)行操作，我這里的columns=range(0,2)，此處是對(duì)第0，2列進(jìn)行操作
  return pd.Series([row[0]**2,row[1]**2],index=['1_1','1_2'])

def f2_2(row1,row2):
  #對(duì)某兩列進(jìn)行操作，我這里的columns=range(0,2)，此處是對(duì)第0，2列進(jìn)行操作
  return pd.Series([row1**2,row2**2],index=['2_1','2_2'])

def apply_f(df):
  return df.apply(f,axis=1)

def apply_f1_1(df):
  return df.apply(f1_1,axis=1)

def apply_f1_2(df):
  return df[0].apply(f1_2)

def apply_f2_1(df):
  return df.apply(f2_1,axis=1)

def apply_f2_2(df):
  return df.apply(lambda row :f2_2(row[0],row[1]),axis=1)
 
def init_process(global_vars):
  global a
  a = global_vars
  
def time_compare():
  '''直接調(diào)用和多線程調(diào)用時(shí)間對(duì)比'''
  a = 2
  np.random.seed(0)
  df = pd.DataFrame(np.random.rand(10**5,2),columns=range(0,2))
  print(df.columns)
   
  t1= time.time()
  result_serial = df.apply(f,axis=1)
  t2 = time.time()
  print("Serial time =",t2-t1)
  print(result_serial.head())

  
  df_parts=np.array_split(df,20)
  print(len(df_parts),type(df_parts[0]))
  with Pool(processes=8,initializer=init_process,initargs=(a,)) as pool: 
  #with Pool(processes=8) as pool:    
    result_parts = pool.map(apply_f,df_parts)
  result_parallel= pd.concat(result_parts)
  t3 = time.time()
  print("Parallel time =",t3-t2)
  print(result_parallel.head())


def apply_fun():
  '''多種apply函數(shù)的調(diào)用'''
  a = 2
  np.random.seed(0)
  df = pd.DataFrame(np.random.rand(10**5,2),columns=range(0,2))
  print(df.columns)
  df_parts=np.array_split(df,20)
  print(len(df_parts),type(df_parts[0]))
  with Pool(processes=8,initializer=init_process,initargs=(a,)) as pool: 
  #with Pool(processes=8) as pool:    
    res_part0 = pool.map(apply_f,df_parts)
    res_part1 = pool.map(apply_f1_1,df_parts)
    res_part2 = pool.map(apply_f1_2,df_parts)
    res_part3 = pool.map(apply_f2_1,df_parts)
    res_part4 = pool.map(apply_f2_2,df_parts)

  res_parallel0 = pd.concat(res_part0)
  res_parallel1 = pd.concat(res_part1)
  res_parallel2 = pd.concat(res_part2)
  res_parallel3 = pd.concat(res_part3)
  res_parallel4 = pd.concat(res_part4)
  
  print("f:\n",res_parallel0.head())
  print("f1:\n",res_parallel1.head())
  print("f2:\n",res_parallel2.head())
  print("f3:\n",res_parallel3.head())
  print("f4:\n",res_parallel4.head())

  df=pd.concat([df,res_parallel0],axis=1)
  df=pd.concat([df,res_parallel1],axis=1)
  df=pd.concat([df,res_parallel2],axis=1)
  df=pd.concat([df,res_parallel3],axis=1)
  df=pd.concat([df,res_parallel4],axis=1)

  print(df.head())
      
  
if __name__ == '__main__':
  time_compare()
  apply_fun()

參考網(wǎng)址

https://blog.fangzhou.me/posts/20170702-python-parallelism/

https://docs.python.org/3.7/library/multiprocessing.html

到此這篇關(guān)于pandas apply多線程實(shí)現(xiàn)代碼的文章就介紹到這了,更多相關(guān)pandas apply多線程內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: