class_docstr: Generic TFX schema_gen executor.
class_docstr_tok: ["Generic","TFX","schema_gen","executor","."]
docstr: TensorFlow SchemaGen executor entrypoint.
  
  This infers the schema using tensorflow_data_validation on precomputed
  stats.
  
  Args:
    input_dict: Input dict from input key to a list of artifacts, including:
      - 'statistics': A list of 'ExampleStatistics' type.
    output_dict: Output dict from key to a list of artifacts, including:
      - schema: A list of 'Schema' artifact of size one.
    exec_properties: A dict of execution properties, includes:
      - infer_feature_shape: Whether or not to infer the shape of the feature.
      - exclude_splits: Names of splits that will not be taken into
        consideration when auto-generating a schema.
  
  Returns:
    None
docstr_tok: ["TensorFlow","SchemaGen","executor","entrypoint",".","This","infers","the","schema","using","tensorflow_data_validation","on","precomputed","stats",".","Args",":","input_dict",":","Input","dict","from","input","key","to","a","list","of","artifacts",",","including",":","-","'statistics","'",":","A","list","of","'ExampleStatistics","'","type",".","output_dict",":","Output","dict","from","key","to","a","list","of","artifacts",",","including",":","-","schema",":","A","list","of","'Schema","'","artifact","of","size","one",".","exec_properties",":","A","dict","of","execution","properties",",","includes",":","-","infer_feature_shape",":","Whether","or","not","to","infer","the","shape","of","the","feature",".","-","exclude_splits",":","Names","of","splits","that","will","not","be","taken","into","consideration","when","auto-generating","a","schema",".","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:
      """TensorFlow SchemaGen executor entrypoint.
  
      This infers the schema using tensorflow_data_validation on precomputed
      stats.
  
      Args:
        input_dict: Input dict from input key to a list of artifacts, including:
          - 'statistics': A list of 'ExampleStatistics' type.
        output_dict: Output dict from key to a list of artifacts, including:
          - schema: A list of 'Schema' artifact of size one.
        exec_properties: A dict of execution properties, includes:
          - infer_feature_shape: Whether or not to infer the shape of the feature.
          - exclude_splits: Names of splits that will not be taken into
            consideration when auto-generating a schema.
  
      Returns:
        None
      """
      infer_feature_shape = bool(
          exec_properties.get(
              standard_component_specs.INFER_FEATURE_SHAPE_KEY, True
          )
      )
  
      # Load and deserialize exclude splits from execution properties.
      exclude_splits = (
          json_utils.loads(
              exec_properties.get(
                  standard_component_specs.EXCLUDE_SPLITS_KEY, "null"
              )
          )
          or []
      )
      if not isinstance(exclude_splits, list):
          raise ValueError(
              "exclude_splits in execution properties needs to be a "
              "list. Got %s instead." % type(exclude_splits)
          )
  
      # Only one schema is generated for all splits.
      schema = None
      stats_artifact = artifact_utils.get_single_instance(
          input_dict[standard_component_specs.STATISTICS_KEY]
      )
      for split in artifact_utils.decode_split_names(
          stats_artifact.split_names
      ):
          if split in exclude_splits:
              continue
          logging.info(
              "Processing schema from statistics for split %s.", split
          )
          stats = stats_artifact_utils.load_statistics(
              stats_artifact, split
          ).proto()
          if not schema:
              schema = tfdv.infer_schema(stats, infer_feature_shape)
          else:
              schema = tfdv.update_schema(
                  schema, stats, infer_feature_shape
              )
      if schema is None:
          raise ValueError(
              "No input splits for stats artifact: %s" % stats_artifact
          )
  
      output_uri = os.path.join(
          artifact_utils.get_single_uri(
              output_dict[standard_component_specs.SCHEMA_KEY]
          ),
          DEFAULT_FILE_NAME,
      )
      io_utils.write_pbtxt_file(output_uri, schema)
      logging.info("Schema written to %s.", output_uri)
  
file_path: tfx/tfx/components/schema_gen/executor.py
imports: import os
  from typing import Any, Dict, List
  from absl import logging
  import tensorflow_data_validation
  from tfx import types
  from tfx.components.statistics_gen import stats_artifact_utils
  from tfx.dsl.components.base import base_executor
  from tfx.types import artifact_utils
  from tfx.types import standard_component_specs
  from tfx.utils import io_utils
  from tfx.utils import json_utils
label_desc: Use image node_id 1 for calling the Executor obj's underlying member method code with example usage: obj.Do(input_dict, output_dict, exec_properties) without return types