当前位置: 首页 > 编程笔记 >

pandas apply多线程实现代码

邵逸明
2023-03-14
本文向大家介绍pandas apply多线程实现代码,包括了pandas apply多线程实现代码的使用技巧和注意事项,需要的朋友参考一下

一、多线程化选择

     并行化一个代码有两大选择:multithread 和 multiprocess。

     Multithread,多线程,同一个进程(process)可以开启多个线程执行计算。每个线程代表了一个 CPU 核心,这么多线程可以访问同样的内存地址(所谓共享内存),实现了线程之间的通讯,算是最简单的并行模型。

    Multiprocess,多进程,则相当于同时开启多个 Python 解释器,每个解释器有自己独有的数据,自然不会有数据冲突。

二、并行化思想

并行化的基本思路是把 dataframe 用 np.array_split 方法切割成多个子 dataframe。再调用 Pool.map 函数并行地执行。注意到顺序执行的 pandas.DataFrame.apply 是如何转化成 Pool.map 然后并行执行的。

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 关键词是 context manager,避免写很繁琐的处理开关进程的逻辑

 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多线程内容请搜索小牛知识库以前的文章或继续浏览下面的相关文章希望大家以后多多支持小牛知识库!

 类似资料:
  • 本文向大家介绍Python实现多线程下载文件的代码实例,包括了Python实现多线程下载文件的代码实例的使用技巧和注意事项,需要的朋友参考一下 实现简单的多线程下载,需要关注如下几点: 1.文件的大小:可以从reponse header中提取,如“Content-Length:911”表示大小是911字节 2.任务拆分:指定各个线程下载的文件的哪一块,可以通过request header中添加“R

  • 本文向大家介绍C#实现多线程的Web代理服务器实例,包括了C#实现多线程的Web代理服务器实例的使用技巧和注意事项,需要的朋友参考一下 本文实例讲述了C#实现多线程的Web代理服务器。分享给大家供大家参考。具体如下: 希望本文所述对大家的C#程序设计有所帮助。

  • 本文向大家介绍python实现多线程采集的2个代码例子,包括了python实现多线程采集的2个代码例子的使用技巧和注意事项,需要的朋友参考一下 代码一: 代码二:

  • 本文向大家介绍Python实现线程池代码分享,包括了Python实现线程池代码分享的使用技巧和注意事项,需要的朋友参考一下 原理:建立一个任务队列,然多个线程都从这个任务队列中取出任务然后执行,当然任务队列要加锁,详细请看代码

  • 本文向大家介绍实例讲解php实现多线程,包括了实例讲解php实现多线程的使用技巧和注意事项,需要的朋友参考一下 我们首先来看一个示例 当执行上述操作时,即使在耗时的处理结束之前也显示处理完成。 由于“ > / dev / null& ”返回的值丢失,我们不会等待响应。 因此,处理完成指示比耗时处理更快。 但是,该方法不知道耗时的处理何时结束。 不要忘记考虑采取日志的执行,登录DB等方法向用户传递执

  • 本文向大家介绍用python实现的线程池实例代码,包括了用python实现的线程池实例代码的使用技巧和注意事项,需要的朋友参考一下 python3标准库里自带线程池ThreadPoolExecutor和进程池ProcessPoolExecutor。 如果你用的是python2,那可以下载一个模块,叫threadpool,这是线程池。对于进程池可以使用python自带的multiprocessing