将异步函数应用于Pandas数据框的最快方法

Question

pandas数据帧中有一个apply方法，可以应用一些同步函数，例如：

import numpy as np
import pandas as pd

def fun(x):
    return x * 2

df = pd.DataFrame(np.arange(10), columns=['old'])

df['new'] = df['old'].apply(fun)

如果有一个需要应用异步函数fun2的类似操作，最快的方法是什么:

import asyncio
import numpy as np
import pandas as pd

async def fun2(x):
    return x * 2

async def main():
    df = pd.DataFrame(np.arange(10), columns=['old'])
    df['new'] = 0    
    for i in range(len(df)):
        df['new'].iloc[i] = await fun2(df['old'].iloc[i])
    print(df)

asyncio.run(main())

- Роман Коптев

Answer 1

使用asyncio.gather，并在完成后覆盖整个列。

import asyncio

import numpy as np
import pandas as pd


async def fun2(x):
    return x * 2


async def main():
    df = pd.DataFrame(np.arange(10), columns=['old'])
    df['new'] = await asyncio.gather(*(fun2(v) for v in df['old']))
    print(df)


asyncio.run(main())

这种方式会将每个列中的值传递给异步函数，这意味着所有列值都将同时运行（比在循环中按顺序等待每个函数结果要快得多）。

注意：通过asyncio.gather保证了列顺序的保留，并且在所有awaitables成功完成之前，该列将不会被解析。

所得到的输出数据框：

   old  new
0    0    0
1    1    2
2    2    4
3    3    6
4    4    8
5    5   10
6    6   12
7    7   14
8    8   16
9    9   18

- Henry Ecker

---

我们如何将输出转换为Pandas数据框？ - snow

@snow 输出是 fun2() 在 df['old'] Series 上的结果列表。@Henry 的答案只是将该列表分配给原始 df 中的新列（Series）。 - Joey Baruch

这是我见过的将pandas与asyncio混合使用的最巧妙的方法之一。实际上，我不得不在浏览器历史记录中搜索它。哈哈 - Joey Baruch

元素的正确顺序是否总是会被保留？ - Bulat Ibragimov

@BulatIbragimov 是的。从链接的收集文档中可以看到：“结果值的顺序对应于_aws_中可等待对象的顺序。” - Henry Ecker