python - 在python中并行化for循环

Question

我有一个字典，其中每个键(日期)都包含一个表(多个格式列表[day1, val11, val21], [day2, va12, val22], [day3, val13, val23], . ... 我想将其转换为 DataFrame；这是通过以下代码完成的:

df4 = pd.DataFrame(columns=sorted(set_days))

for date in dic.keys():
        days = [day  for day, val1, val2  in dic[date]]
        val1 = [val1 for day, val1, val2  in dic[date]]
        df4.loc[date, days] = val1

这段代码工作正常，但运行需要两个多小时。经过一些研究，我意识到我可以通过 multiprocessing 库将其并行化；以下代码是预期的并行版本

import multiprocessing

def func(date):
    global df4, dic
    days = [day  for day, val1, val2  in dic[date]]
    val1 = [val1 for day, val1, val2  in dic[date]]
    df4.loc[date, days] = val1

multiprocessing.Pool(processes=8).map(func, dic.keys())

此代码的问题在于，在执行 multiprocessing.Pool(processes... 之后，df4 DataFrame 为空。

如有任何帮助，我们将不胜感激。

示例

假设字典包含两天:

dic['20030812'][:4]
Out: [[1, 24.25, 0.0], [20, 23.54, 23.54], [30, 23.13, 24.36], [50, 22.85, 23.57]]

dic['20030813'][:4]
Out: [[1, 24.23, 0.0], [19, 23.4, 22.82], [30, 22.97, 24.19], [49, 22.74, 23.25]]

那么 DataFrame 应该是这样的形式:

df4.loc[:, 1:50]
             1    2    3    4    5   ...     46   47   48     49     50
20030812  24.25  NaN  NaN  NaN  NaN  ...    NaN  NaN  NaN    NaN  22.85
20030813  24.23  NaN  NaN  NaN  NaN  ...    NaN  NaN  NaN  22.74    NaN

此外，

dic.keys()
Out[36]: dict_keys(['20030812', '20030813'])

df1.head().to_dict()
Out: 
{1: {'20030812': 24.25, '20030813': 24.23},
 2: {'20030812': nan, '20030813': nan},
 3: {'20030812': nan, '20030813': nan},
 4: {'20030812': nan, '20030813': nan},
 5: {'20030812': nan, '20030813': nan},
 6: {'20030812': nan, '20030813': nan},
 7: {'20030812': nan, '20030813': nan},
 8: {'20030812': nan, '20030813': nan},
 9: {'20030812': nan, '20030813': nan},
 10: {'20030812': nan, '20030813': nan},
 11: {'20030812': nan, '20030813': nan},
 12: {'20030812': nan, '20030813': nan},
 13: {'20030812': nan, '20030813': nan},
 14: {'20030812': nan, '20030813': nan},
 15: {'20030812': nan, '20030813': nan},
 16: {'20030812': nan, '20030813': nan},
 17: {'20030812': nan, '20030813': nan},
 18: {'20030812': nan, '20030813': nan},
 19: {'20030812': nan, '20030813': 23.4},
 20: {'20030812': 23.54, '20030813': nan},
 21: {'20030812': nan, '20030813': nan},
 22: {'20030812': nan, '20030813': nan},
 23: {'20030812': nan, '20030813': nan},
 24: {'20030812': nan, '20030813': nan},
 25: {'20030812': nan, '20030813': nan},
 26: {'20030812': nan, '20030813': nan},
 27: {'20030812': nan, '20030813': nan},
 28: {'20030812': nan, '20030813': nan},
 29: {'20030812': nan, '20030813': nan},
 30: {'20030812': 23.13, '20030813': 22.97},
 31: {'20030812': nan, '20030813': nan},
 32: {'20030812': nan, '20030813': nan},
 ...

Answer 1

要回答您最初的问题(大致是:“为什么 df4 DataFrame 是空的？”)，这不起作用的原因是当 Pool 工作人员启动后，每个 worker 都会继承父数据的个人写时复制 View (如果 multiprocessing 在具有 fork 的类 UNIX 系统上运行，则直接继承，或者通过在 Windows 上运行时模拟它的笨拙方法)。

因此，当每个 worker 做:

 df4.loc[date, days] = val1

它正在改变 worker 个人的 df4 副本；父进程的副本保持不变。

一般来说，有3种处理方式:

1. 将您的工作函数更改为返回可在父进程中使用的内容。例如，与其尝试使用 df4.loc[date, days] = val1 执行就地突变，不如返回在父级中执行此操作所需的内容，例如返回日期、天数、val1，然后将父级更改为:

   for date, days, val in multiprocessing.Pool(processes=8).map(func, dic.keys()):
       df4.loc[date, days] = val
   

   这种方法的缺点是它需要对每个返回值进行 pickle(Python 的序列化版本)，从子级通过管道传输到父级，然后取消 pickled；如果 worker 任务不做太多工作，特别是如果返回值很大(在这种情况下，似乎就是这种情况)，它很容易在序列化和 IPC 上花费比在并行性上获得的时间更多的时间。

2. 使用共享对象/内存(在 this answer to "Multiprocessing writing to pandas dataframe" 中演示)。在实践中，这通常不会给你带来太多好处，因为不是基于 the more "raw" ctypes sharing using multiprocessing.sharedctypes 的东西。最终仍然需要将数据从一个进程传输到另一个进程；不过，基于 sharedctypes 的东西可以获得有意义的速度提升，因为一旦映射，共享原始 C 数组的访问速度几乎与本地内存一样快。

3. 如果并行化的工作受 I/O 限制，或使用第三方 C 扩展来处理受 CPU 限制的工作(例如 numpy)，您可能能够从中获得所需的速度提升线程，尽管有 GIL 干扰，线程确实共享相同的内存。您的情况似乎既没有受到 I/O 限制，也没有明显依赖于可能会释放 GIL 的第三方 C 扩展，因此它在这里可能无济于事，但总的来说，从基于进程的并行性切换的简单方法到基于线程的并行性(当您已经在使用 multiprocessing 时)是将 import 从:

   import multiprocessing
   

   到

   import multiprocessing.dummy as multiprocessing
   

   进口the thread-backed version of multiprocessing在预期的名称下，因此代码从使用进程无缝切换到线程。