并行化 DataFrame 自定义函数 Dask答案

【问题标题】：Parallelized DataFrame Custom Function Dask并行化 DataFrame 自定义函数 Dask
【发布时间】：2020-10-13 20:36:21
【问题描述】：

我正在尝试使用 Dask 通过 Dask 的多处理功能加速 Python DataFrame 以进行循环操作。我完全意识到 for-looping 数据帧通常不是最佳实践，但在我的情况下，它是必需的。我已经大量阅读了文档和其他类似问题，但我似乎无法弄清楚我的问题。

df.head()
         Title                                                                                                                                       Content
0  Lizzibtz     @Ontario2020 @Travisdhanraj @fordnation Maybe.  They are not adding to the stress of education during Covid. Texas sample.  Plus…  
1  Jess ????????????️‍????  @BetoORourke So ashamed at how Abbott has not handled COVID in Texas. A majority of our large cities are hot spots with no end in sight.    
2  sidi diallo  New post (PVC Working Gloves) has been published on Covid-19 News Info - Texas test                    
3  Kautillya    @PandaJay What was the need to go to SC for yatra anyway? Isn't covid cases spiking exponentially? Ambubachi mela o… texas
4  SarahLou♡    RT @BenJolly9: 23rd June 2020 was the day Sir Keir Starmer let the Tories off the hook for their miss-handling of COVID-19. texas

我有一个自定义 python 函数定义为：

def locMp(df):
    hitList = []
    for i in range(len(df)):
        print(i)
        string = df.iloc[i]['Content']
        # print(string)
        doc = nlp(string)
        ents = [e.text for e in doc.ents if e.label_ == "GPE"]
        x = np.array(ents)
        print(np.unique(x))
        hitList.append(np.unique(x))

    df['Locations'] = hitList
    return df

此函数添加了从名为 spacy 的库中提取的位置数据框列 - 我认为这并不重要，但我希望您看到整个函数。

现在，通过文档和其他一些问题。将 Dask 的多处理用于数据帧的方法是创建一个 Dask 数据帧，对其进行分区，map_partitions 和 .compute()。因此，我尝试了以下方法和其他一些方法，但都没有成功：

part = 7
ddf = dd.from_pandas(df, npartitions=part)
location = ddf.map_partitions(lambda df: df.apply(locMp), meta=pd.DataFrame).compute()

# and...

part = 7
ddf = dd.from_pandas(df, npartitions=part)
location = ddf.map_partitions(locMp, meta=pd.DataFrame).compute()

# and simplifying from Dask documentation

part = 7
ddf = dd.from_pandas(df, npartitions=part)
location = ddf.map_partitions(locMp)

我用dask.delayed 尝试了其他一些方法，但似乎没有任何效果。我要么得到一个 Dask 系列或其他一些不想要的输出，要么该函数花费的时间与定期运行它一样长或更长。如何使用 Dask 加速自定义 DataFrame 函数操作并返回干净的 Pandas Dataframe？

谢谢

【问题讨论】：

您介意提供mcve 吗？特别是（至少）原始df的样本？
查看编辑 - 带有字符串 Title 和字符串 content 的简单数据框。为了便于测试，我将 Texas 添加到每一行。
要运行实际的库，您可能需要在代码中python -m spacy download en_core_web_sm 然后nlp = en_core_web_sm.load()。这应该允许该函数实际识别位置
您介意分享您尝试的错误吗？也许df.head().to_dict() 的输出也会很棒。您对此df.head() 的预期输出是什么，这将有助于改进功能。

标签： python pandas dataframe dask

【解决方案1】：

您可以尝试让 Dask 处理应用程序，而不是自己进行循环：

ddf["Locations"] = ddf["Content"].apply(
    lambda string: [e.text for e in nlp(string).ents if e.label_ == "GPE"],
    meta=("Content", "object"))

【讨论】：

嗯，我绝对喜欢这个主意。我现在不是工作，所以我无法测试。如果可行，我会接受。
告诉我进展如何。我在一个测试活页夹中启动了它，它看起来对我来说很好，但我不得不删除 nlp 函数所以我不确定。