pandas apply多线程实现代码
本文详解如何在Python中使用多线程加速pandas的apply操作。通过对比多线程与多进程，介绍利用multiprocessing.Pool进行数据分割与并行执行的方法，包含进程初始化、共享变量设置及完整代码示例，帮助提升数据处理效率。
多线程化选择
一、多线程化选择
对代码进行并行化主要有两种路径：多线程（multithread）和多进程（multiprocess）。
多线程模式下，同一个进程可以创建多个线程来执行计算任务。每个线程对应一个CPU核心，它们共享同一块内存地址，从而实现线程间的通信，这是最基础的并行模型。
多进程模式则相当于同时启动多个Python解释器实例，每个解释器拥有独立的数据空间，因此天然避免了数据冲突问题。
并行化思想
二、并行化思想
并行化的核心思路是将DataFrame通过np.array_split方法拆分为多个子DataFrame，随后调用Pool.map函数执行并行计算。这一过程展示了如何将顺序执行的pandas.DataFrame.apply转化为Pool.map并实现并行处理。
Pool类与进程初始化
Pool对象代表一组并行运行的进程，相关开源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())
设置进程初始化函数
def init_process(global_vars):
global a
a = global_vars
设置进程初始化函数
Pool(processes=8,initializer=init_process,initargs=(a,))
此处指定生成8个进程，每个进程启动时需运行init_process函数，其参数为一个包含singleton tuple的元组a。借助init_process和initargs参数，我们可以便捷地设定需要在进程间共享的全局变量（即变量a）。
使用with关键词作为上下文管理器，可以简化进程开关处理的繁琐逻辑。
with Pool(processes=8,initializer=init_process,initargs=(a,)) as pool:
result_parts = pool.map(apply_f,df_parts)
多线程化应用
三、多线程化应用
多线程时间比较及多线程在几种apply场景中的应用
import numpy as np
import pandas as pd
import time
from multiprocessing import Pool
def f(row):
#直接对某列进行操作
return sum(row)+a
def f1_1(row):
#对某一列进行操作，我这里的columns=range(0,2)，此处是对第0列进行操作
return row[0]**2
def f1_2(row1):
#对某一列进行操作，我这里的columns=range(0,2)，此处是对第0列进行操作
return row1**2
def f2_1(row):
#对某两列进行操作，我这里的columns=range(0,2)，此处是对第0，2列进行操作
return pd.Series([row[0]**2,row[1]**2],index=['1_1','1_2'])
def f2_2(row1,row2):
#对某两列进行操作，我这里的columns=range(0,2)，此处是对第0，2列进行操作
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():
'''直接调用和多线程调用时间对比'''
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函数的调用'''
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()参考网址
https://blog.fangzhou.me/posts/20170702-python-parallelism/
https://docs.python.org/3.7/library/multiprocessing.html
到此这篇关于pandas apply多线程实现代码的文章就介绍到这了,更多相关pandas apply多线程内容请搜索本站以前的文章或继续浏览下面的相关文章希望大家以后多多支持本站！
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
