class_docstr: MetaflowTask prepares a Flow instance for execution of a single step.
class_docstr_tok: ["MetaflowTask","prepares","a","Flow","instance","for","execution","of","a","single","step","."]
code: def _init_data(self, run_id, join_type, input_paths):
      # We prefer to use the parallelized version to initialize datastores
      # (via TaskDataStoreSet) only with more than 4 datastores, because
      # the baseline overhead of using the set is ~1.5s and each datastore
      # init takes ~200-300ms when run sequentially.
      if len(input_paths) > 4:
          prefetch_data_artifacts = None
          if join_type and join_type == "foreach":
              # Prefetch 'foreach' related artifacts to improve time taken by
              # _init_foreach.
              prefetch_data_artifacts = [
                  "_foreach_stack",
                  "_foreach_num_splits",
                  "_foreach_var",
              ]
          # Note: Specify `pathspecs` while creating the datastore set to
          # guarantee strong consistency and guard against missing input.
          datastore_set = TaskDataStoreSet(
              self.flow_datastore,
              run_id,
              pathspecs=input_paths,
              prefetch_data_artifacts=prefetch_data_artifacts,
          )
          ds_list = [ds for ds in datastore_set]
          if len(ds_list) != len(input_paths):
              raise MetaflowDataMissing(
                  "Some input datastores are missing. "
                  "Expected: %d Actual: %d"
                  % (len(input_paths), len(ds_list))
              )
      else:
          # initialize directly in the single input case.
          ds_list = []
          for input_path in input_paths:
              run_id, step_name, task_id = input_path.split("/")
              ds_list.append(
                  self.flow_datastore.get_task_datastore(
                      run_id, step_name, task_id
                  )
              )
      if not ds_list:
          # this guards against errors in input paths
          raise MetaflowDataMissing(
              "Input paths *%s* resolved to zero inputs"
              % ",".join(input_paths)
          )
      return ds_list
  
imports: from __future__ import print_function
  from io import BytesIO
  import math
  import sys
  import os
  import time
  from types import MethodType, FunctionType
  from metaflow.datastore.exceptions import DataException
  from .metaflow_config import MAX_ATTEMPTS
  from .metadata import MetaDatum
  from .mflog import TASK_LOG_SOURCE
  from .datastore import Inputs, TaskDataStoreSet
  from .exception import MetaflowInternalError, MetaflowDataMissing, MetaflowExceptionWrapper
  from .unbounded_foreach import UBF_CONTROL
  from .util import all_equal, get_username, resolve_identity, unicode_type
  from .current import current
  from metaflow.tracing import get_trace_id
  from collections import namedtuple
label_desc: Use image node_id 4 for calling the MetaflowTask obj's underlying member method code with example usage: obj._init_data(run_id, join_type, input_paths) and returns: ds_list