class_docstr: Util class to build splits.
  
  Usage is as follow:
  
  ```py
  split_builder = SplitBuilder(...)
  
  with split_builder.maybe_beam_pipeline():
    split_info_future = split_builder.submit_split_generation(...)
  
  split_info = split_info_future.result()
  ```
  
  * submit_split_generation:
    * For generator based split: Generate the split
    * For Apache Beam based split: Create the `beam.PCollection` and returns
      a future.
  * `split_info_future.result()`: Called after all `beam.PCollection`s have
    finished. Finalize the `split_info` by collecting all pipeline results.
  
  `submit_split_generation` / `.result` should be called once per
  split.
class_docstr_tok: ["Util","class","to","build","splits",".","Usage","is","as","follow",":","``","`","py","split_builder","=","SplitBuilder","(","...",")","with","split_builder.maybe_beam_pipeline","(",")",":","split_info_future","=","split_builder.submit_split_generation","(","...",")","split_info","=","split_info_future.result","(",")","``","`","*","submit_split_generation",":","*","For","generator","based","split",":","Generate","the","split","*","For","Apache","Beam","based","split",":","Create","the","`","beam.PCollection","`","and","returns","a","future",".","*","`","split_info_future.result","(",")","`",":","Called","after","all","`","beam.PCollection","`","s","have","finished",".","Finalize","the","`","split_info","`","by","collecting","all","pipeline","results",".","`","submit_split_generation","`","\/","`",".result","`","should","be","called","once","per","split","."]
docstr: Split generator for `beam.PCollection`.
docstr_tok: ["Split","generator","for","`","beam.PCollection","`","."]
code: def _build_from_pcollection(
      self,
      split_name: str,
      generator: "beam.PCollection[KeyExample]",
      filename_template: naming.ShardedFileTemplate,
      disable_shuffling: bool,
  ) -> _SplitInfoFuture:
      """Split generator for `beam.PCollection`."""
      # TODO(tfds): Should try to add support to `max_examples_per_split`
      beam_writer = writer_lib.BeamWriter(
          serializer=example_serializer.ExampleSerializer(
              self._features.get_serialized_info()
          ),
          filename_template=filename_template,
          hash_salt=split_name,
          disable_shuffling=disable_shuffling,
          file_format=self._file_format,
          shard_config=self._shard_config,
      )
  
      def _encode_example(
          key_ex, encode_fn=self._features.encode_example
      ):
          # We do not access self._features in this function to avoid pickling the
          # entire class.
          return key_ex[0], encode_fn(key_ex[1])
  
      # Note: We need to wrap the pipeline in a PTransform to avoid
      # errors due to duplicated ``>> beam_labels`
      @beam.ptransform_fn
      def _encode_pcollection(pipeline):
          """PTransformation which build a single split."""
          pcoll_examples = pipeline | "Encode" >> beam.Map(
              _encode_example
          )
          return beam_writer.write_from_pcollection(pcoll_examples)
  
      # Add the PCollection to the pipeline
      _ = (
          generator | f"{split_name}_write" >> _encode_pcollection()
      )  # pylint: disable=no-value-for-parameter
  
      def _resolve_future():
          if self._in_contextmanager:
              raise AssertionError(
                  "`future.result()` should be called after the "
                  "`maybe_beam_pipeline` contextmanager."
              )
          logging.info("Retrieving split info for %s...", split_name)
          shard_lengths, total_size = beam_writer.finalize()
          return splits_lib.SplitInfo(
              name=split_name,
              shard_lengths=shard_lengths,
              num_bytes=total_size,
              filename_template=filename_template,
          )
  
      return _SplitInfoFuture(_resolve_future)
  
imports: import collections.abc
  import contextlib
  import dataclasses
  import functools
  import itertools
  import sys
  from typing import Any, Callable, Dict, Iterable, Iterator, List, Optional, Tuple, Union
  from absl import logging
  import click
  import psutil
  from tensorflow_datasets.core import example_serializer
  from tensorflow_datasets.core import features
  from tensorflow_datasets.core import file_adapters
  from tensorflow_datasets.core import naming
  from tensorflow_datasets.core import splits
  from tensorflow_datasets.core import utils
  from tensorflow_datasets.core import writer
  from tensorflow_datasets.core.utils import shard_utils
  from tensorflow_datasets.core.utils.lazy_imports_utils import apache_beam
label_desc: Use image node_id 7 for calling the SplitBuilder obj's underlying member method code with example usage: obj._build_from_pcollection(split_name, generator, filename_template, disable_shuffling) and returns: _SplitInfoFuture, key_ex,  encode_fn, beam_writer, splits_lib