code: def beam_import(
      dataset,
      samples,
      parse_fcn=None,
      expand_schema=True,
      dynamic=False,
      validate=True,
      options=None,
      verbose=False,
  ):
      """Imports the given samples into the dataset via
      `Apache Beam <https://beam.apache.org>`_.
  
      This function is a parallelized alternative to
      :meth:`fiftyone.core.dataset.Dataset.add_samples`.
  
      .. note::
  
          The insertion order of the samples is not guaranteed.
  
      Example::
  
          import fiftyone as fo
          import fiftyone.utils.beam as foub
  
          samples = range(10000)
  
          def make_sample(idx):
              return fo.Sample(filepath="image%d.png" % idx, uuid=idx)
  
          #
          # Option 1: build the samples on the workers
          #
  
          dataset = fo.Dataset()
  
          foub.beam_import(dataset, samples, parse_fcn=make_sample)
          print(dataset)
  
          #
          # Option 2: build the samples in the main thread
          #
          # This is generally not preferred but may be necessary if your
          # ``parse_fcn`` is not serializable
          #
  
          dataset = fo.Dataset()
  
          samples = map(make_sample, samples)
  
          foub.beam_import(dataset, samples)
          print(dataset)
  
      Args:
          dataset: a :class:`fiftyone.core.dataset.Dataset`
          samples: an iterable of samples. If no ``parse_fcn`` is provided, these
              must be :class:`fiftyone.core.sample.Sample` instances. If a
              ``parse_fcn`` is provided, these are passed to it for parsing
          parse_fcn (None): an optional function that converts elements of
              ``samples`` to :class:`fiftyone.core.sample.Sample` instances
          expand_schema (True): whether to dynamically add new sample fields
              encountered to the dataset schema. If False, an error is raised
              if a sample's schema is not a subset of the dataset schema
          dynamic (False): whether to declare dynamic attributes of embedded
              document fields that are encountered
          validate (True): whether to validate that the fields of each sample
              are compliant with the dataset schema before adding it
          options (None): a
              ``apache_beam.options.pipeline_options.PipelineOptions`` that
              configures how to run the pipeline. By default, the pipeline will
              be run via Beam's direct runner using threads
          verbose (False): whether to log the Beam pipeline's messages
      """
      if options is None:
          options = PipelineOptions(
              runner="direct",
              direct_num_workers=fou.recommend_thread_pool_workers(),
              direct_running_mode="multi_threading",
          )
  
      sample0, samples = _pop_first(samples)
  
      if sample0 is None:
          return  # empty
  
      if parse_fcn is not None:
          sample0 = parse_fcn(sample0)
  
      # Manually insert first sample to reduce chances of parallel schema changes
      dataset.add_sample(
          sample0,
          expand_schema=expand_schema,
          dynamic=dynamic,
          validate=validate,
      )
  
      if parse_fcn is None:
          # `Sample` objects are not serializable so we must manually serialize
          # and deserialize in `ImportBatch`
          samples = map(
              lambda s: s.to_mongo_dict(include_id=True), samples
          )
  
      import_batch = ImportBatch(
          dataset.name,
          parse_fcn=parse_fcn,
          expand_schema=expand_schema,
          dynamic=dynamic,
          validate=validate,
      )
  
      logger = logging.getLogger()
      level = logger.level if verbose else logging.CRITICAL
      with fou.SetAttributes(logger, level=level):
          with beam.Pipeline(options=options) as pipeline:
              _ = (
                  pipeline
                  | "InitImport" >> beam.Create(samples)
                  | "ImportBatches" >> beam.ParDo(import_batch)
              )
  
      # The dataset's schema may have changed in another process
      dataset.reload()
  
imports: import logging
  import numpy
  import fiftyone.core.dataset
  import fiftyone.core.utils
  import fiftyone.core.view
  import apache_beam
  from apache_beam.options.pipeline_options import PipelineOptions
label_desc: Use image node_id 1 for calling a global function with example usage: beam_import(dataset, samples, parse_fcn, expand_schema, dynamic, validate, options, verbose) without return types