code: def parallel_calculate_chunks(
      cutoff_time,
      chunk_size,
      feature_set,
      approximate,
      training_window,
      save_progress,
      entityset,
      n_jobs,
      no_unapproximated_aggs,
      cutoff_df_time_col,
      target_time,
      pass_columns,
      progress_bar,
      dask_kwargs=None,
      progress_callback=None,
      include_cutoff_time=True,
  ):
      import_or_raise(
          "distributed",
          "Dask must be installed to calculate feature matrix with n_jobs set to anything but 1",
      )
      from dask.base import tokenize
      from distributed import Future, as_completed
  
      client = None
      cluster = None
      try:
          client, cluster = create_client_and_cluster(
              n_jobs=n_jobs,
              dask_kwargs=dask_kwargs,
              entityset_size=entityset.__sizeof__(),
          )
          # scatter the entityset
          # denote future with leading underscore
          start = time.time()
          es_token = "EntitySet-{}".format(tokenize(entityset))
          if es_token in client.list_datasets():
              msg = "Using EntitySet persisted on the cluster as dataset {}"
              progress_bar.write(msg.format(es_token))
              _es = client.get_dataset(es_token)
          else:
              _es = client.scatter([entityset])[0]
              client.publish_dataset(**{_es.key: _es})
  
          # save features to a tempfile and scatter it
          pickled_feats = cloudpickle.dumps(feature_set)
          _saved_features = client.scatter(pickled_feats)
          client.replicate([_es, _saved_features])
          num_scattered_workers = len(
              client.who_has([Future(es_token)]).get(es_token, []),
          )
          num_workers = len(client.scheduler_info()["workers"].values())
  
          schema = None
          if isinstance(cutoff_time, pd.DataFrame):
              schema = cutoff_time.ww.schema
              chunks = cutoff_time.groupby(cutoff_df_time_col)
              cutoff_time_len = cutoff_time.shape[0]
          else:
              chunks = cutoff_time
              cutoff_time_len = len(cutoff_time[1])
  
          if not chunk_size:
              chunk_size = _handle_chunk_size(
                  1.0 / num_workers, cutoff_time_len
              )
  
          chunks = _chunk_dataframe_groups(chunks, chunk_size)
  
          chunks = [df for _, df in chunks]
  
          if len(chunks) < num_workers:  # pragma: no cover
              chunk_warning = "Fewer chunks ({}), than workers ({}) consider reducing the chunk size"
              warning_string = chunk_warning.format(
                  len(chunks), num_workers
              )
              progress_bar.write(warning_string)
  
          scatter_warning(num_scattered_workers, num_workers)
          end = time.time()
          scatter_time = round(end - start)
  
          # if enabled, reset timer after scatter for better time remaining estimates
          if not progress_bar.disable:
              progress_bar.reset()
  
          scatter_string = (
              "EntitySet scattered to {} workers in {} seconds"
          )
          progress_bar.write(
              scatter_string.format(num_scattered_workers, scatter_time)
          )
          # map chunks
          # TODO: consider handling task submission dask kwargs
          _chunks = client.map(
              calculate_chunk,
              chunks,
              feature_set=_saved_features,
              chunk_size=None,
              entityset=_es,
              approximate=approximate,
              training_window=training_window,
              save_progress=save_progress,
              no_unapproximated_aggs=no_unapproximated_aggs,
              cutoff_df_time_col=cutoff_df_time_col,
              target_time=target_time,
              pass_columns=pass_columns,
              progress_bar=None,
              progress_callback=progress_callback,
              include_cutoff_time=include_cutoff_time,
              schema=schema,
          )
  
          feature_matrix = []
          iterator = as_completed(_chunks).batches()
          for batch in iterator:
              results = client.gather(batch)
              for result in results:
                  feature_matrix.append(result)
                  previous_progress = progress_bar.n
                  progress_bar.update(result.shape[0])
                  if progress_callback is not None:
                      (
                          update,
                          progress_percent,
                          time_elapsed,
                      ) = update_progress_callback_parameters(
                          progress_bar,
                          previous_progress,
                      )
                      progress_callback(
                          update, progress_percent, time_elapsed
                      )
  
      except Exception:
          raise
      finally:
          if client is not None:
              client.close()
  
          if "cluster" not in dask_kwargs and cluster is not None:
              cluster.close()  # pragma: no cover
  
      ww_init_kwargs = get_ww_types_from_features(
          feature_set.target_features,
          entityset,
          pass_columns,
          cutoff_time,
      )
      feature_matrix = init_ww_and_concat_fm(
          feature_matrix, ww_init_kwargs
      )
      return feature_matrix
  
imports: import logging
  import math
  import os
  import shutil
  import time
  import warnings
  from datetime import datetime
  import cloudpickle
  import numpy
  import pandas
  from woodwork.logical_types import Age, AgeNullable, Boolean, BooleanNullable, Integer, IntegerNullable
  from featuretools.computational_backends.feature_set import FeatureSet
  from featuretools.computational_backends.feature_set_calculator import FeatureSetCalculator
  from featuretools.computational_backends.utils import _check_cutoff_time_type, _validate_cutoff_time, bin_cutoff_times, create_client_and_cluster, gather_approximate_features, gen_empty_approx_features_df, get_ww_types_from_features, save_csv_decorator
  from featuretools.entityset.relationship import RelationshipPath
  from featuretools.feature_base import AggregationFeature, FeatureBase
  from featuretools.utils import Trie
  from featuretools.utils.gen_utils import Library, import_or_none, import_or_raise, is_instance, make_tqdm_iterator
label_desc: Use image node_id 5 for calling a global function with example usage: parallel_calculate_chunks(cutoff_time, chunk_size, feature_set, approximate, training_window, save_progress, entityset, n_jobs, no_unapproximated_aggs, cutoff_df_time_col, target_time, pass_columns, progress_bar, dask_kwargs, progress_callback, include_cutoff_time) and returns: feature_matrix