code: def beam_merge(
      dataset,
      samples,
      parse_fcn=None,
      options=None,
      verbose=False,
      **kwargs,
  ):
      """Merges 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.merge_samples`.
  
      .. note::
  
          This function is only useful for merging **in-memory samples** into a
          dataset. If you are merging a sample collection, simply call
          :meth:`fiftyone.core.dataset.Dataset.merge_samples`.
  
      Example::
  
          import fiftyone as fo
          import fiftyone.utils.beam as foub
          import fiftyone.zoo as foz
  
          dataset = foz.load_zoo_dataset("quickstart").clone()
  
          samples = iter(dataset.select_fields("predictions"))
  
          foub.beam_merge(dataset, samples, fields={"predictions": "predictions2"})
  
          print(dataset.count("predictions.detections"))
          print(dataset.count("predictions2.detections"))
  
      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
          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
          **kwargs: keyword arguments for
              :meth:`fiftyone.core.dataset.Dataset.merge_samples`
      """
  
      # If the merge does not require a `key_fcn`, then it is fastest to import
      # the samples into a temporary collection and then merge that
      if kwargs.get("key_fcn", None) is None:
          tmp_dataset = fod.Dataset()
  
          try:
              beam_import(
                  tmp_dataset,
                  samples,
                  parse_fcn=parse_fcn,
                  options=options,
                  verbose=verbose,
              )
  
              dataset.merge_samples(tmp_dataset, **kwargs)
          finally:
              tmp_dataset.delete()
  
          return
  
      if options is None:
          options = PipelineOptions(
              runner="direct",
              direct_num_workers=fou.recommend_thread_pool_workers(),
              direct_running_mode="multi_threading",
          )
  
      if parse_fcn is None:
          # `Sample` objects are not serializable so we must manually serialize
          # and deserialize in `MergeBatch`
          samples = map(
              lambda s: s.to_mongo_dict(include_id=True), samples
          )
  
      merge_batch = MergeBatch(
          dataset.name, parse_fcn=parse_fcn, **kwargs
      )
  
      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
                  | "InitMerge" >> beam.Create(samples)
                  | "MergeBatches" >> beam.ParDo(merge_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 2 for calling a global function with example usage: beam_merge(dataset, samples, parse_fcn, options, verbose) without return types