class_docstr: Generic TFX example gen base executor.
  
  The base ExampleGen executor takes a configuration and converts external data
  sources to TensorFlow Examples (tf.train.Example, tf.train.SequenceExample),
  or any other protocol buffer as subclass defines.
  
  The common configuration (defined in
  https://github.com/tensorflow/tfx/blob/master/tfx/proto/example_gen.proto#L44.)
  describes the general properties of input data and shared instructions when
  producing output data.
  
  The conversion is done in `GenerateExamplesByBeam` as a Beam pipeline, which
  validates the configuration, reads the external data sources, converts the
  record in the input source to any supported output payload formats
  (e.g., tf.Example or tf.SequenceExample) if needed, and splits the examples
  if the output split config is given. Then the executor's `Do` writes the
  results in splits to the output path.
  
  For simple custom ExampleGens, the details of transforming input data
  record(s) to a specific output payload format (e.g., tf.Example or
  tf.SequenceExample) is expected to be given in
  `GetInputSourceToExamplePTransform`, which returns a Beam PTransform with the
  actual implementation. For complex use cases, such as joining multiple data
  sources and different interpretations of the configurations, the custom
  ExampleGen can override `GenerateExamplesByBeam`.
docstr: Converts input source to serialized record splits based on configs.
  
  Custom ExampleGen executor should provide GetInputSourceToExamplePTransform
  for converting input split to serialized records. Overriding this
  'GenerateExamplesByBeam' method instead if complex logic is need, e.g.,
  custom spliting logic.
  
  Args:
    pipeline: Beam pipeline.
    exec_properties: A dict of execution properties. Depends on detailed
      example gen implementation.
      - input_base: an external directory containing the data files.
      - input_config: JSON string of example_gen_pb2.Input instance, providing
        input configuration.
      - output_config: JSON string of example_gen_pb2.Output instance,
        providing output configuration.
      - output_data_format: Payload format of generated data in output
        artifact, one of example_gen_pb2.PayloadFormat enum.
  
  Returns:
    Dict of beam PCollection with split name as key, each PCollection is a
    single output split that contains serialized records.
docstr_tok: ["Converts","input","source","to","serialized","record","splits","based","on","configs",".","Custom","ExampleGen","executor","should","provide","GetInputSourceToExamplePTransform","for","converting","input","split","to","serialized","records",".","Overriding","this","'GenerateExamplesByBeam","'","method","instead","if","complex","logic","is","need",",","e.g.",",","custom","spliting","logic",".","Args",":","pipeline",":","Beam","pipeline",".","exec_properties",":","A","dict","of","execution","properties",".","Depends","on","detailed","example","gen","implementation",".","-","input_base",":","an","external","directory","containing","the","data","files",".","-","input_config",":","JSON","string","of","example_gen_pb2.Input","instance",",","providing","input","configuration",".","-","output_config",":","JSON","string","of","example_gen_pb2.Output","instance",",","providing","output","configuration",".","-","output_data_format",":","Payload","format","of","generated","data","in","output","artifact",",","one","of","example_gen_pb2.PayloadFormat","enum",".","Returns",":","Dict","of","beam","PCollection","with","split","name","as","key",",","each","PCollection","is","a","single","output","split","that","contains","serialized","records","."]
code: def GenerateExamplesByBeam(
      self,
      pipeline: beam.Pipeline,
      exec_properties: Dict[str, Any],
  ) -> Dict[str, beam.pvalue.PCollection]:
      """Converts input source to serialized record splits based on configs.
  
      Custom ExampleGen executor should provide GetInputSourceToExamplePTransform
      for converting input split to serialized records. Overriding this
      'GenerateExamplesByBeam' method instead if complex logic is need, e.g.,
      custom spliting logic.
  
      Args:
        pipeline: Beam pipeline.
        exec_properties: A dict of execution properties. Depends on detailed
          example gen implementation.
          - input_base: an external directory containing the data files.
          - input_config: JSON string of example_gen_pb2.Input instance, providing
            input configuration.
          - output_config: JSON string of example_gen_pb2.Output instance,
            providing output configuration.
          - output_data_format: Payload format of generated data in output
            artifact, one of example_gen_pb2.PayloadFormat enum.
  
      Returns:
        Dict of beam PCollection with split name as key, each PCollection is a
        single output split that contains serialized records.
      """
      # Get input split information.
      input_config = example_gen_pb2.Input()
      proto_utils.json_to_proto(
          exec_properties[standard_component_specs.INPUT_CONFIG_KEY],
          input_config,
      )
      # Get output split information.
      output_config = example_gen_pb2.Output()
      proto_utils.json_to_proto(
          exec_properties[standard_component_specs.OUTPUT_CONFIG_KEY],
          output_config,
      )
      # Get output split names.
      split_names = utils.generate_output_split_names(
          input_config, output_config
      )
      # Make beam_pipeline_args available in exec_properties since certain
      # example_gen executors need this information.
      exec_properties["_beam_pipeline_args"] = (
          self._beam_pipeline_args or []
      )
  
      example_splits = []
      input_to_record = self.GetInputSourceToExamplePTransform()
      if output_config.split_config.splits:
          # Use output splits, input must have only one split.
          assert (
              len(input_config.splits) == 1
          ), "input must have only one split when output split is specified."
          # Calculate split buckets.
          buckets = []
          total_buckets = 0
          for split in output_config.split_config.splits:
              total_buckets += split.hash_buckets
              buckets.append(total_buckets)
          example_splits = (
              pipeline
              | "InputToRecord" >>
              # pylint: disable=no-value-for-parameter
              input_to_record(
                  exec_properties, input_config.splits[0].pattern
              )
              | "SplitData"
              >> beam.Partition(
                  _PartitionFn,
                  len(buckets),
                  buckets,
                  output_config.split_config,
              )
          )
      else:
          # Use input splits.
          for split in input_config.splits:
              examples = (
                  pipeline
                  | "InputToRecord[{}]".format(split.name) >>
                  # pylint: disable=no-value-for-parameter
                  input_to_record(exec_properties, split.pattern)
              )
              example_splits.append(examples)
  
      result = {}
      for index, example_split in enumerate(example_splits):
          result[split_names[index]] = example_split
      return result
  
imports: import abc
  import bisect
  import hashlib
  import pickle
  from typing import Any, Dict, List, Union
  from absl import logging
  import apache_beam
  import tensorflow
  from tfx import types
  from tfx.components.example_gen import utils
  from tfx.components.example_gen import write_split
  from tfx.components.util import examples_utils
  from tfx.dsl.components.base import base_beam_executor
  from tfx.proto import example_gen_pb2
  from tfx.types import artifact_utils
  from tfx.types import standard_component_specs
  from tfx.utils import proto_utils
label_desc: Use image node_id 2 for calling the BaseExampleGenExecutor obj's underlying member method code with example usage: obj.GenerateExamplesByBeam(pipeline, exec_properties) and returns: result