class_docstr: Convert a collection of text documents to a matrix of token counts
  
  Notes
  -----
  When a vocabulary isn't provided, ``fit_transform`` requires two
  passes over the dataset: one to learn the vocabulary and a second
  to transform the data. Consider persisting the data if it fits
  in (distributed) memory prior to calling ``fit`` or ``transform``
  when not providing a ``vocabulary``.
  
  Additionally, this implementation benefits from having
  an active ``dask.distributed.Client``, even on a single machine.
  When a client is present, the learned ``vocabulary`` is persisted
  in distributed memory, which saves some recomputation and redundant
  communication.
  
  See Also
  --------
  sklearn.feature_extraction.text.CountVectorizer
  
  Examples
  --------
  The Dask-ML implementation currently requires that ``raw_documents``
  is a :class:`dask.bag.Bag` of documents (lists of strings).
  
  >>> from dask_ml.feature_extraction.text import CountVectorizer
  >>> import dask.bag as db
  >>> from distributed import Client
  >>> client = Client()
  >>> corpus = [
  ...     'This is the first document.',
  ...     'This document is the second document.',
  ...     'And this is the third one.',
  ...     'Is this the first document?',
  ... ]
  >>> corpus = db.from_sequence(corpus, npartitions=2)
  >>> vectorizer = CountVectorizer()
  >>> X = vectorizer.fit_transform(corpus)
  dask.array<concatenate, shape=(nan, 9), dtype=int64, chunksize=(nan, 9), ...
             chunktype=scipy.csr_matrix>
  >>> X.compute().toarray()
  array([[0, 1, 1, 1, 0, 0, 1, 0, 1],
         [0, 2, 0, 1, 0, 1, 1, 0, 1],
         [1, 0, 0, 1, 1, 0, 1, 1, 1],
         [0, 1, 1, 1, 0, 0, 1, 0, 1]])
  >>> vectorizer.get_feature_names()
  ['and', 'document', 'first', 'is', 'one', 'second', 'the', 'third', 'this']
class_docstr_tok: ["Convert","a","collection","of","text","documents","to","a","matrix","of","token","counts","Notes","--","--","-","When","a","vocabulary","is","n't","provided",",","``","fit_transform","``","requires","two","passes","over","the","dataset",":","one","to","learn","the","vocabulary","and","a","second","to","transform","the","data",".","Consider","persisting","the","data","if","it","fits","in","(","distributed",")","memory","prior","to","calling","``","fit","``","or","``","transform","``","when","not","providing","a","``","vocabulary","``",".","Additionally",",","this","implementation","benefits","from","having","an","active","``","dask.distributed.Client","``",",","even","on","a","single","machine",".","When","a","client","is","present",",","the","learned","``","vocabulary","``","is","persisted","in","distributed","memory",",","which","saves","some","recomputation","and","redundant","communication",".","See","Also","--","--","--","--","sklearn.feature_extraction.text.CountVectorizer","Examples","--","--","--","--","The","Dask-ML","implementation","currently","requires","that","``","raw_documents","``","is","a",":","class",":","`","dask.bag.Bag","`","of","documents","(","lists","of","strings",")",".",">",">",">","from","dask_ml.feature_extraction.text","import","CountVectorizer",">",">",">","import","dask.bag","as","db",">",">",">","from","distributed","import","Client",">",">",">","client","=","Client","(",")",">",">",">","corpus","=","[","...","'This","is","the","first","document",".","'",",","...","'This","document","is","the","second","document",".","'",",","...","'And","this","is","the","third","one",".","'",",","...","'Is","this","the","first","document","?","'",",","...","]",">",">",">","corpus","=","db.from_sequence","(","corpus",",","npartitions=2",")",">",">",">","vectorizer","=","CountVectorizer","(",")",">",">",">","X","=","vectorizer.fit_transform","(","corpus",")","dask.array","<","concatenate",",","shape=","(","nan",",","9",")",",","dtype=int64",",","chunksize=","(","nan",",","9",")",",","...","chunktype=scipy.csr_matrix",">",">",">",">","X.compute","(",")",".toarray","(",")","array","(","[","[","0",",","1",",","1",",","1",",","0",",","0",",","1",",","0",",","1","]",",","[","0",",","2",",","0",",","1",",","0",",","1",",","1",",","0",",","1","]",",","[","1",",","0",",","0",",","1",",","1",",","0",",","1",",","1",",","1","]",",","[","0",",","1",",","1",",","1",",","0",",","0",",","1",",","0",",","1","]","]",")",">",">",">","vectorizer.get_feature_names","(",")","[","'and","'",",","'document","'",",","'first","'",",","'is","'",",","'one","'",",","'second","'",",","'the","'",",","'third","'",",","'this","'","]"]
code: def transform(self, raw_documents):
      params = self.get_params()
      vocabulary = params.pop("vocabulary")
  
      if vocabulary is None:
          check_is_fitted(self, "vocabulary_")
          vocabulary = self.vocabulary_
  
      if isinstance(vocabulary, dict):
          # scatter for the user
          try:
              client = get_client()
          except ValueError:
              vocabulary_for_transform = dask.delayed(vocabulary)
          else:
              (vocabulary_for_transform,) = client.scatter(
                  (vocabulary,), broadcast=True
              )
      else:
          vocabulary_for_transform = vocabulary
  
      n_features = vocabulary_length(vocabulary_for_transform)
      transformed = raw_documents.map_partitions(
          _count_vectorizer_transform, vocabulary_for_transform, params
      )
      meta = scipy.sparse.eye(0, format="csr", dtype=self.dtype)
      return build_array(transformed, n_features, meta)
  
imports: import itertools
  import dask
  import dask.array
  import dask.bag
  import dask.dataframe
  import distributed
  import numpy
  import scipy.sparse
  import sklearn.base
  import sklearn.feature_extraction.text
  from dask.delayed import Delayed
  from distributed import get_client, wait
  from sklearn.utils.validation import check_is_fitted
label_desc: Use image node_id 2 for calling the CountVectorizer obj's underlying member method code with example usage: obj.transform(raw_documents) and returns: build_array