[Python] How to Apply an Async Function to Every Row in a Dask DataFrame?

I have a Dask DataFrame with multiple partitions, and I need to apply an async function to every row. However, I’m encountering issues because apply() is not designed to work with async functions.

Here’s a simplified version of what I’m trying to do:

import lorem
import pandas as pd
import dask.dataframe as dd

# Example async function
async def async_api_call(text: str):
return text[:3]

# Create a dataset with 100 rows of text
df = pd.DataFrame({"Text": [lorem.text() for _ in range(100)]})
ddf = dd.from_pandas(df, npartitions=4)

# Attempting to apply the async function
ddf["new"] = ddf.map_partitions(
lambda partition: partition["Text"].apply(async_api_call),
meta=(None, str),
)


Running this simply returns the unexecuted coroutines, here is a screenshot:



questions:

• What is the correct way to apply an async function to each row in a Dask DataFrame?
• Can I make use of something like asyncio.run() or another workaround?

Continue reading...