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: Take input data source and generates serialized data splits.
  
  The output is intended to be serialized tf.train.Examples or
  tf.train.SequenceExamples protocol buffer in gzipped TFRecord format,
  but subclasses can choose to override to write to any serialized records
  payload into gzipped TFRecord as specified, so long as downstream
  component can consume it. The format of payload is added to
  `payload_format` custom property of the output Example artifact.
  
  Args:
    input_dict: Input dict from input key to a list of Artifacts. Depends on
      detailed example gen implementation.
    output_dict: Output dict from output key to a list of Artifacts.
      - examples: splits of serialized records.
    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.
      - output_file_format: File format of generated data in output artifact,
        one of example_gen_pb2.FileFormat enum.
  
  Returns:
    None
docstr_tok: ["Take","input","data","source","and","generates","serialized","data","splits",".","The","output","is","intended","to","be","serialized","tf.train.Examples","or","tf.train.SequenceExamples","protocol","buffer","in","gzipped","TFRecord","format",",","but","subclasses","can","choose","to","override","to","write","to","any","serialized","records","payload","into","gzipped","TFRecord","as","specified",",","so","long","as","downstream","component","can","consume","it",".","The","format","of","payload","is","added","to","`","payload_format","`","custom","property","of","the","output","Example","artifact",".","Args",":","input_dict",":","Input","dict","from","input","key","to","a","list","of","Artifacts",".","Depends","on","detailed","example","gen","implementation",".","output_dict",":","Output","dict","from","output","key","to","a","list","of","Artifacts",".","-","examples",":","splits","of","serialized","records",".","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",".","-","output_file_format",":","File","format","of","generated","data","in","output","artifact",",","one","of","example_gen_pb2.FileFormat","enum",".","Returns",":","None"]
code: def Do(
      self,
      input_dict: Dict[str, List[types.Artifact]],
      output_dict: Dict[str, List[types.Artifact]],
      exec_properties: Dict[str, Any],
  ) -> None:
      """Take input data source and generates serialized data splits.
  
      The output is intended to be serialized tf.train.Examples or
      tf.train.SequenceExamples protocol buffer in gzipped TFRecord format,
      but subclasses can choose to override to write to any serialized records
      payload into gzipped TFRecord as specified, so long as downstream
      component can consume it. The format of payload is added to
      `payload_format` custom property of the output Example artifact.
  
      Args:
        input_dict: Input dict from input key to a list of Artifacts. Depends on
          detailed example gen implementation.
        output_dict: Output dict from output key to a list of Artifacts.
          - examples: splits of serialized records.
        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.
          - output_file_format: File format of generated data in output artifact,
            one of example_gen_pb2.FileFormat enum.
  
      Returns:
        None
      """
      self._log_startup(input_dict, output_dict, exec_properties)
  
      input_config = example_gen_pb2.Input()
      proto_utils.json_to_proto(
          exec_properties[standard_component_specs.INPUT_CONFIG_KEY],
          input_config,
      )
      output_config = example_gen_pb2.Output()
      proto_utils.json_to_proto(
          exec_properties[standard_component_specs.OUTPUT_CONFIG_KEY],
          output_config,
      )
  
      examples_artifact = artifact_utils.get_single_instance(
          output_dict[standard_component_specs.EXAMPLES_KEY]
      )
      examples_artifact.split_names = artifact_utils.encode_split_names(
          utils.generate_output_split_names(input_config, output_config)
      )
  
      output_file_format = exec_properties.get(
          standard_component_specs.OUTPUT_FILE_FORMAT_KEY,
          example_gen_pb2.FILE_FORMAT_UNSPECIFIED,
      )
  
      logging.info("Generating examples.")
      with self._make_beam_pipeline() as pipeline:
          example_splits = self.GenerateExamplesByBeam(
              pipeline, exec_properties
          )
  
          # pylint: disable=expression-not-assigned, no-value-for-parameter
          for split_name, example_split in example_splits.items():
              (
                  example_split
                  | "WriteSplit[{}]".format(split_name)
                  >> write_split.WriteSplit(
                      artifact_utils.get_split_uri(
                          output_dict[
                              standard_component_specs.EXAMPLES_KEY
                          ],
                          split_name,
                      ),
                      output_file_format,
                      exec_properties,
                  )
              )
          # pylint: enable=expression-not-assigned, no-value-for-parameter
  
      output_payload_format = exec_properties.get(
          standard_component_specs.OUTPUT_DATA_FORMAT_KEY
      )
      if output_payload_format:
          for output_examples_artifact in output_dict[
              standard_component_specs.EXAMPLES_KEY
          ]:
              examples_utils.set_payload_format(
                  output_examples_artifact, output_payload_format
              )
  
      if output_file_format:
          for output_examples_artifact in output_dict[
              standard_component_specs.EXAMPLES_KEY
          ]:
              examples_utils.set_file_format(
                  output_examples_artifact,
                  write_split.to_file_format_str(output_file_format),
              )
  
      logging.info("Examples generated.")
  
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 3 for calling the BaseExampleGenExecutor obj's underlying member method code with example usage: obj.Do(input_dict, output_dict, exec_properties) without return types