当我尝试使用 HashingVectorizer 时出现 Dask client.persist returns AssertionError

Question

我正在尝试使用 dask HashingVectorizer 对 dask.dataframe 进行矢量化。我希望矢量化结果保留在集群（分布式系统）中。这就是我在尝试转换数据时使用 client.persist 的原因。但由于某种原因，我收到以下错误。

Traceback (most recent call last):
  File "/home/dodzilla/my_project/components_with_adapter/vectorizers/base_vectorizer.py", line 112, in hybrid_feature_vectorizer
    CLUSTERING_FEATURES=self.clustering_features)
  File "/home/dodzilla/my_project/components_with_adapter/vectorizers/text_vectorizer.py", line 143, in vectorize
    X = self.client.persist(fitted_vectorizer.transform, combined_data)
  File "/home/dodzilla/.local/lib/python3.6/site-packages/distributed/client.py", line 2860, in persist
    assert all(map(dask.is_dask_collection, collections))
AssertionError

我无法共享数据，但有关数据的所有必要信息如下：

>>>type(combined_data)
<class 'dask.dataframe.core.Series'>
>>>type(combined_data.compute())
<class 'pandas.core.series.Series'>
>>>combined_data.compute().shape
12

可以在下面找到一个最小的工作示例。下面的代码片段中，combined_data 包含合并的列。含义：所有列合并为 1 列。数据有 12 行。行内的所有值都是字符串。这是我收到错误的代码：

from stop_words import get_stop_words
from dask_ml.feature_extraction.text import HashingVectorizer as daskHashingVectorizer
import pandas as pd
import dask
import dask.dataframe as dd
from dask.distributed import Client


def convert_dataframe_to_single_text(documents):
    """
    Combine all of the columns into 1 column.
    """
    if type(documents) is dask.dataframe.core.DataFrame:
        cols = documents.columns
        documents['combined'] = documents[cols].apply(func=(lambda row: ' '.join(row.values.astype(str))), axis=1,
                                                      meta=('str'))
        document_texts = documents.drop(cols, axis=1)
    else:
        raise TypeError('Wrong type of data. Expected Pandas DF or Dask DF but received ', type(documents))
    return document_texts

# Init the client.
client = Client('localhost:8786')

# Get stopwords
stopwords = get_stop_words(language="english")

# Create dask dataframe from pandas dataframe
data = {'Name':['Tom', 'nick', 'krish', 'jack'], 'Age':["twenty", "twentyone", "nineteen", "eighteen"]}
df = pd.DataFrame(data)
df = dd.from_pandas(df, npartitions=1)

# Init the vectorizer
vectorizer = daskHashingVectorizer(stop_words=stopwords, alternate_sign=False,
                       norm=None, binary=False,
                       n_features=10000)

# Combine all of to columns into 1 column.
combined_data = convert_dataframe_to_single_text(df)

# Fit the vectorizer.
fitted_vectorizer = client.persist(vectorizer.fit(combined_data))

# Transform the data.
X = client.persist(fitted_vectorizer.transform, combined_data)

希望信息足够了

重要说明： 当我说 client.compute 时我没有收到任何类型的错误，但据我所知，这在机器集群中不起作用，并且相反，它在本地机器上运行。它 returns 是一个 csr 矩阵而不是延迟计算的 dask.array.

Answer 1

这不是我应该使用的方式 client.persist。我正在寻找的功能是 client.submit 和 client.map... 在我的情况下 client.submit 解决了我的问题。

当我尝试使用 HashingVectorizer 时出现 Dask client.persist returns AssertionError

Dask client.persist returns AssertionError when I try to use HashingVectorizer

python-3.x

dask

dask-distributed