code: def distribute(self, func, partitioned_chunks, kwargs):
      def _wrapped_func(*args, **kw):
          # Series.value_counts() may not be able to handle
          if not getattr(pd.Series.value_counts, "_wrapped", False):
              old_value_counts = pd.Series.value_counts
  
              def _wrapped_value_counts(obj, *args, **kw):
                  try:
                      return old_value_counts(obj, *args, **kw)
                  except ValueError:  # pragma: no cover
                      return old_value_counts(obj.copy(), *args, **kw)
  
              pd.Series.value_counts = _wrapped_value_counts
              pd.Series.value_counts._wrapped = True
  
          return func(*args, **kw)
  
      tasks = []
      for partitioned_chunk in partitioned_chunks:
          tasks.append(
              mr.spawn(
                  _wrapped_func,
                  args=(partitioned_chunk,),
                  kwargs=kwargs,
              )
          )
      executed = mr.ExecutableTuple(tasks).execute(
          session=self._session
      )
      fetched = executed.fetch(session=self._session)
      return [item for results in fetched for item in results]
  
file_path: mars/mars/learn/contrib/tsfresh/core.py
imports: import pandas
  from ....None import remote
  from ....session import get_default_session
  from ....utils import ceildiv
label_desc: Use image node_id 3 for calling the MarsDistributor obj's underlying member method code with example usage: obj.distribute(func, partitioned_chunks, kwargs) and returns: unknown, func, old_value_counts, old_value_counts