イントロ

現在、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']))]

リストコンプ私はこのソリューションをテストしていません

その他のパンダソリューション

tagged_document = list(read_corpus_tag_sub(df))

このソリューションは、何時間にもわたって動きます。しかし、それが終わったときにこのことをやりくりするのに十分な記憶がありません。

結論（？）

私は今スーパーロストを感じています。これが私が見たスレッドのリストです。私は本当に多くの時間を過ごしたばかりで、ばかげた用事をしているような気がします。

1. ジェネレーターのDaskBag
2. Daskでテキストを処理する
3. Daskを使用してパンダの適用をスピードアップ
4. 1台のマシンですべてのコアを利用してPandasデータフレームでapply（）をどのように並列化しますか？
5. python dask DataFrame、（自明に並列化可能な）行のサポートが適用されますか？
6. map_partitionsは何をしていますか？
7. 単純なdaskmap_partitionsの例
8. ドキュメント

私はDaskAPI /制限に精通していませんが、一般的には次のとおりです。

• データを（単語、タグ）タプルとして繰り返すことができれば（Doc2Vec/TaggedDocumentステップを無視しても）、Dask側が処理され、それらのタプルをTaggedDocumentインスタンスに変換するのは簡単です。

• 大規模なデータセットのために、一般的に、あなたはしたくない（とするのに十分なRAMを持っていない場合があります）として、完全なデータセットをインスタンス化listのメモリに-あなたの関与の試みそうlist()や.append()ポイントまで、作業していて、ローカルメモリを排出することができます（深刻なスワッピングを引き起こす）および/またはデータの終わりに達していない。

大規模なデータセットへの好ましいアプローチは、データを反復するように求められるたびに（Doc2Vecトレーニングには複数のパスが必要になるため）、すべてのアイテムを順番に提供できる反復可能なオブジェクトを作成することですが、データセット全体をインメモリオブジェクト。

このパターンに関する優れたブログ投稿は次のとおりです。Pythonでのデータストリーミング：ジェネレーター、イテレーター、イテレーター

あなたが示したコードを考えると、あなたにとって正しいアプローチは次のようになると思います。

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

そうです、あなたはこのコードに近づいています

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バッグパンダ/ DASK DATAFRAMEよりも、より自然リストや発電機などのハンドルことを行い、。