python - 将 dask 数据框中的列转换为 Doc2Vec 的 TaggedDocument

Question

简介

目前，我正在尝试使用 dask 与 gensim 配合进行 NLP 文档计算，但在将我的语料库转换为“TaggedDocument”时遇到了问题。

因为我尝试了很多不同的方法来解决这个问题，所以我将列出我的尝试。

处理这个问题的每次尝试都会遇到略有不同的困境。

首先是一些初始给定。

数据

df.info()
<class 'dask.dataframe.core.DataFrame'>
Columns: 5 entries, claim_no to litigation
dtypes: object(2), int64(3)

  claim_no   claim_txt I                                    CL ICC lit
0 8697278-17 battery comprising interior battery active ele... 106 2 0

期望的输出

>>tagged_document[0]
>>TaggedDocument(words=['battery', 'comprising', 'interior', 'battery', 'active', 'elements', 'battery', 'cell', 'casing', 'said', 'cell', 'casing', 'comprising', 'first', 'casing', 'element', 'first', 'contact', 'surface', 'second', 'casing', 'element', 'second', 'contact', 'surface', 'wherein', 'assembled', 'position', 'first', 'second', 'contact', 'surfaces', 'contact', 'first', 'second', 'casing', 'elements', 'encase', 'active', 'materials', 'battery', 'cell', 'interior', 'space', 'wherein', 'least', 'one', 'gas', 'tight', 'seal', 'layer', 'arranged', 'first', 'second', 'contact', 'surfaces', 'seal', 'interior', 'space', 'characterized', 'one', 'first', 'second', 'contact', 'surfaces', 'comprises', 'electrically', 'insulating', 'void', 'volume', 'layer', 'first', 'second', 'contact', 'surfaces', 'comprises', 'formable', 'material', 'layer', 'fills', 'voids', 'surface', 'void', 'volume', 'layer', 'hermetically', 'assembled', 'position', 'form', 'seal', 'layer'], tags=['8697278-17'])
>>len(tagged_document) == len(df['claim_txt'])

错误号 1 不允许生成器

def read_corpus_tag_sub(df,corp='claim_txt',tags=['claim_no']):
    for i, line in enumerate(df[corp]):
        yield gensim.models.doc2vec.TaggedDocument(gensim.utils.simple_preprocess(line), (list(df.loc[i,tags].values)))

tagged_document = df.map_partitions(read_corpus_tag_sub,meta=TaggedDocument)
tagged_document = tagged_document.compute()

类型错误:无法序列化类型生成器的对象。

我发现在仍然使用发电机的情况下没有办法解决这个问题。解决这个问题就太好了!因为这对于普通的 Pandas 来说非常有效。

错误号 2 仅每个分区的第一个元素

def read_corpus_tag_sub(df,corp='claim_txt',tags=['claim_no']):
    for i, line in enumerate(df[corp]):
        return gensim.models.doc2vec.TaggedDocument(gensim.utils.simple_preprocess(line), (list(df.loc[i,tags].values)))

tagged_document = df.map_partitions(read_corpus_tag_sub,meta=TaggedDocument)
tagged_document = tagged_document.compute()

这个有点愚蠢，因为该函数不会迭代(我知道)，但会给出所需的格式，但只返回每个分区中的第一行。

错误号 3 函数调用因 100% cpu 而挂起

def read_corpus_tag_sub(df,corp='claim_txt',tags=['claim_no']):
    tagged_list = []
    for i, line in enumerate(df[corp]):
        tagged = gensim.models.doc2vec.TaggedDocument(gensim.utils.simple_preprocess(line), (list(df.loc[i,tags].values)))
        tagged_list.append(tagged)
    return tagged_list

据我所知，在循环外部重构返回时，该函数会挂起 dask 客户端中的构建内存，并且我的 CPU 利用率达到 100%，但没有计算任何任务。请记住，我以相同的方式调用该函数。

Pandas 解决方案

def tag_corp(corp,tag):
    return gensim.models.doc2vec.TaggedDocument(gensim.utils.simple_preprocess(corp), ([tag]))

tagged_document = [tag_corp(x,y) for x,y in list(zip(df_smple['claim_txt'],df_smple['claim_no']))]

列出比较我还没有测试过这个解决方案

其他 Pandas 解决方案

tagged_document = list(read_corpus_tag_sub(df))

这个解决方案将持续几个小时。然而，当它完成后，我没有足够的内存来处理这件事。

结论(？)

我现在感觉 super 迷失。这是我看过的主题列表。我承认我对 dask 还很陌生，我刚刚花了很多时间，但我觉得我在做一件愚蠢的事。

1. Dask Bag from generator
2. Processing Text With Dask
3. Speed up Pandas apply using Dask
4. How do you parallelize apply() on Pandas Dataframes making use of all cores on one machine?
5. python dask DataFrame, support for (trivially parallelizable) row apply?
6. What is map_partitions doing?
7. simple dask map_partitions example
8. The Docs

Answer 1

我不熟悉 Dask API/限制，但一般来说:

• 如果您可以将数据作为(单词、标签)​​元组进行迭代 - 甚至忽略 Doc2Vec/TaggedDocument 步骤 - 那么 Dask 端将是处理，并将这些元组转换为 TaggedDocument 实例应该很简单

• 一般来说，对于大型数据集，您不想(并且可能没有足够的 RAM 来)将完整数据集实例化为内存中的列表 - 因此您的尝试涉及list() 或 .append() 可能在一定程度上有效，但会耗尽本地内存(导致严重交换)和/或只是没有到达您的末尾数据。

处理大型数据集的最佳方法是创建一个可迭代对象，每次要求迭代数据时(因为 Doc2Vec 训练需要多次传递)，该对象都可以提供每个数据集依次项 - 但永远不会将整个数据集读取到内存中的对象中。

关于此模式的一篇很好的博文是:Data streaming in Python: generators, iterators, iterables

鉴于您所显示的代码，我怀疑适合您的方法可能是这样的:

from gensim.utils import simple_preprocess

class MyDataframeCorpus(object):
    def __init__(self, source_df, text_col, tag_col):
        self.source_df = source_df
        self.text_col = text_col
        self.tag_col = tag_col

    def __iter__(self):
        for i, row in self.source_df.iterrows():
            yield TaggedDocument(words=simple_preprocess(row[self.text_col]), 
                                 tags=[row[self.tag_col]])

corpus_for_doc2vec = MyDataframeCorpus(df, 'claim_txt', 'claim_no')