将 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()

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

我发现在仍然使用生成器的同时无法解决这个问题。解决这个问题会很棒！因为这对于普通熊猫来说非常有效。

错误号 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%，但没有计算任何任务。请记住，我以同样的方式调用函数。

熊猫解决方案

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']))]

List comp 我没有时间测试这个解决方案

其他 Pandas 解决方案

tagged_document = list(read_corpus_tag_sub(df))

这个解决方案会持续好几个小时。但是我没有足够的内存来处理这件事。

结论(?)

我现在感觉超级迷茫。这是我看过的线程列表。我承认我对 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/限制，但通常：

1234563实例应该是微不足道的

• 一般而言，对于大型数据集，您不希望（并且可能没有足够的 RAM 来）将整个数据集实例化为内存中的 list - 所以您的尝试涉及 list() 或 @ 987654327@ 可能在某种程度上可以工作，但会耗尽本地内存（导致严重的交换）和/或只是没有到达数据的末尾。

对于大型数据集，最好的方法是创建一个可迭代对象，每次要求它迭代数据时（因为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')

Answer 2

好的，所以你很接近这段代码

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)

但正如您所见，生成生成器对 Dask 并没有多大帮助。相反，你可以让你的函数返回一个系列

def myfunc(df, *args, **kwargs):
    output = []
    for i, line in enumerate(df["my_series"])
        result = ...
        output.append([])
    return pd.Series(output)

或者，您可能只使用df.apply 方法，该方法采用将单行转换为单行的函数。

您可能还想切换到 Dask Bag，它确实比 Pandas/Dask DataFrame 更自然地处理列表和生成器之类的事情。