pandas apply 函数 实现多进程的示例讲解

# groupby 操作
df1 = pd.DataFrame({'a':[1,2,1,2,1,2], 'b':[3,3,3,4,4,4], 'data':[12,13,11,8,10,3]})
df1

grouped = df1.groupby('b')
# 按照 'b' 这列分组了，name 为 'b' 的 key 值，group 为对应的df_group
for name, group in grouped:
 print name, '->'
 print group

3 ->
 a b data
0 1 3 12
1 2 3 13
2 1 3 11
4 ->
 a b data
3 2 4  8
4 1 4 10
5 2 4  3

grouped = df1.groupby(['a','b'])
# 按照 'b' 这列分组了，name 为 'b' 的 key 值，group 为对应的df_group
for name, group in grouped:
 print name, '->'
 print group

(1, 3) ->
 a b data
0 1 3 12
2 1 3 11
(1, 4) ->
 a b data
4 1 4 10
(2, 3) ->
 a b data
1 2 3 13
(2, 4) ->
 a b data
3 2 4  8
5 2 4  3

grouped = df1.groupby(df1.index)
# 按照 index 分组，其实每行就是一个组了
print len(grouped), type(grouped)
for name, group in grouped:
 print name, '->'
 print group

6 <class 'pandas.core.groupby.DataFrameGroupBy'>
0 ->
 a b data
0 1 3 12
1 ->
 a b data
1 2 3 13
2 ->
 a b data
2 1 3 11
3 ->
 a b data
3 2 4  8
4 ->
 a b data
4 1 4 10
5 ->
 a b data
5 2 4  3

# 1. Embarrassingly parallel helper: to make it easy to write readable parallel code and debug it quickly:
from joblib import Parallel, delayed
from math import sqrt

%time result1 = Parallel(n_jobs=1)(delayed(sqrt)(i**2) for i in range(10000))
%time result2 = Parallel(n_jobs=8)(delayed(sqrt)(i**2) for i in range(10000))

CPU times: user 316 ms, sys: 0 ns, total: 316 ms
Wall time: 309 ms
CPU times: user 692 ms, sys: 384 ms, total: 1.08 s
Wall time: 1.03 s

%time result = Parallel(n_jobs=1)(delayed(sqrt)(i**2) for i in range(1000000))

CPU times: user 3min 43s, sys: 5.66 s, total: 3min 49s
Wall time: 3min 33s

%time result = Parallel(n_jobs=8)(delayed(sqrt)(i**2) for i in range(1000000))

CPU times: user 50.9 s, sys: 12.6 s, total: 1min 3s
Wall time: 52 s

# 读入所有停用词
with open('stopwords.txt', 'rb') as inp:
 lines = inp.read()
stopwords = re.findall('"(.*?)"', lines)
print len(stopwords)
print stopwords[:10]

692
['a', "a's", 'able', 'about', 'above', 'according', 'accordingly', 'across', 'actually', 'after']

# 对 AbstractText 去停用词
# 方法一：暴力法，对每个词进行判断
def remove_stopwords1(text):
 words = text.split(' ')
 new_words = list()
 for word in words:
  if word not in stopwords:
   new_words.append(word)
 return new_words
# 方法二：先构建停用词的映射
for word in stopwords:
 if word in words_count.index:
  words_count[word] = -1
def remove_stopwords2(text):
 words = text.split(' ')
 new_words = list()
 for word in words:
  if words_count[word] != -1:
   new_words.append(word)
 return new_words
%time df_all['AbstractText1'] = df_all['AbstractText'].apply(remove_stopwords1)
%time df_all['AbstractText2'] = df_all['AbstractText'].apply(remove_stopwords2)

CPU times: user 8min 56s, sys: 2.72 s, total: 8min 59s
Wall time: 8min 48s
CPU times: user 1min 2s, sys: 4.12 s, total: 1min 6s
Wall time: 1min 2s

from joblib import Parallel, delayed
import multiprocessing
# 方法三：对方法一使用多进程
def tmp_func(df):
 df['AbstractText3'] = df['AbstractText'].apply(remove_stopwords1)
 return df
def apply_parallel(df_grouped, func):
 """利用 Parallel 和 delayed 函数实现并行运算"""
 results = Parallel(n_jobs=-1)(delayed(func)(group) for name, group in df_grouped)
 return pd.concat(results)
if __name__ == '__main__':
 time0 = time.time()
 df_grouped = df_all.groupby(df_all.index)
 df_all =applyParallel(df_grouped, tmp_func)
 print 'time costed {0:.2f}'.format(time.time() - time0)

time costed 150.81

# 方法四：对方法二使用多进程
def tmp_func(df):
 df['AbstractText3'] = df['AbstractText'].apply(remove_stopwords2)
 return df
def apply_parallel(df_grouped, func):
 """利用 Parallel 和 delayed 函数实现并行运算"""
 results = Parallel(n_jobs=-1)(delayed(func)(group) for name, group in df_grouped)
 return pd.concat(results)
if __name__ == '__main__':
 time0 = time.time()
 df_grouped = df_all.groupby(df_all.index)
 df_all =applyParallel(df_grouped, tmp_func)
 print 'time costed {0:.2f}'.format(time.time() - time0)

time costed 123.80