code: def beam_export(
      sample_collection,
      num_shards,
      options=None,
      verbose=False,
      render_kwargs=None,
      **kwargs,
  ):
      """Exports the given sample collection in the specified number shards via
      `Apache Beam <https://beam.apache.org>`_.
  
      This function is a parallelized alternative to
      :meth:`fiftyone.core.collections.SampleCollection.export` that effectively
      performs the following sharded export in parallel::
  
          for idx, (first, last) in enumerate(shards, 1):
              _kwargs = render_kwargs(kwargs, idx)
              sample_collection[first:last].export(**_kwargs)
  
      Example::
  
          from apache_beam.options.pipeline_options import PipelineOptions
  
          import fiftyone as fo
          import fiftyone.utils.beam as foub
          import fiftyone.zoo as foz
  
          dataset = foz.load_zoo_dataset("quickstart")
  
          # Use multithreading instead of the default multiprocessing
          options = PipelineOptions(
              runner="direct",
              direct_num_workers=10,
              direct_running_mode="multi_threading",
          )
  
          foub.beam_export(
              dataset,
              num_shards=20,
              options=options,
              dataset_type=fo.types.TFObjectDetectionDataset,
              label_field="ground_truth",
              tf_records_path="/tmp/beam/tf.records-%05d-of-00020",
          )
  
      Args:
          sample_collection: a
              :class:`fiftyone.core.collections.SampleCollection`
          num_shards: the number of shards to write
          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 multiprocessing
          verbose (False): whether to log the Beam pipeline's messages
          render_kwargs (None): a function that renders ``kwargs`` for the
              current shard. The function should have signature
              ``def render_kwargs(kwargs, idx) -> kwargs``, where ``idx`` in
              ``[1, num_shards]`` is the shard index. By default, any
              string-valued arguments that contain format patterns like ``%05d``
              will be rendered via ``value % idx``
          **kwargs: keyword arguments for
              :meth:`fiftyone.core.collections.SampleCollection.export`
      """
      if options is None:
          num_workers = min(
              num_shards, fou.recommend_process_pool_workers()
          )
          options = PipelineOptions(
              runner="direct",
              direct_num_workers=num_workers,
              direct_running_mode="multi_processing",
          )
  
      if isinstance(sample_collection, fov.DatasetView):
          dataset_name = sample_collection._root_dataset.name
          view_stages = sample_collection._serialize()
      else:
          dataset_name = sample_collection.name
          view_stages = None
  
      export_batch = ExportBatch(
          dataset_name,
          view_stages=view_stages,
          render_kwargs=render_kwargs,
      )
  
      n = len(sample_collection)
      edges = [int(round(b)) for b in np.linspace(0, n, num_shards + 1)]
  
      batches = [
          {"idx": idx, "start": start, "stop": stop}
          for idx, (start, stop) in enumerate(
              zip(edges[:-1], edges[1:]), 1
          )
      ]
  
      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
                  | "InitExport" >> beam.Create(batches)
                  | "ExportBatches"
                  >> beam.ParDo(export_batch, **kwargs)
              )
  
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 3 for calling a global function with example usage: beam_export(sample_collection, num_shards, options, verbose, render_kwargs) without return types