class_docstr: DistributionValidator component executor.
docstr: DistributionValidator executor entrypoint.
  
  This checks for changes in data distributions from one dataset to another,
  based on the summary statitics for those datasets.
  
  Args:
    input_dict: Input dict from input key to a list of artifacts.
    output_dict: Output dict from key to a list of artifacts.
    exec_properties: A dict of execution properties.
  
  Returns:
    ExecutionResult proto with anomalies and the component generated alerts
    execution property set with anomalies alerts, if any.
docstr_tok: ["DistributionValidator","executor","entrypoint",".","This","checks","for","changes","in","data","distributions","from","one","dataset","to","another",",","based","on","the","summary","statitics","for","those","datasets",".","Args",":","input_dict",":","Input","dict","from","input","key","to","a","list","of","artifacts",".","output_dict",":","Output","dict","from","key","to","a","list","of","artifacts",".","exec_properties",":","A","dict","of","execution","properties",".","Returns",":","ExecutionResult","proto","with","anomalies","and","the","component","generated","alerts","execution","property","set","with","anomalies","alerts",",","if","any","."]
code: def Do(
      self,
      input_dict: Dict[str, list[types.Artifact]],
      output_dict: Dict[str, list[types.Artifact]],
      exec_properties: Dict[str, Any],
  ) -> execution_result_pb2.ExecutorOutput:
      """DistributionValidator executor entrypoint.
  
      This checks for changes in data distributions from one dataset to another,
      based on the summary statitics for those datasets.
  
      Args:
        input_dict: Input dict from input key to a list of artifacts.
        output_dict: Output dict from key to a list of artifacts.
        exec_properties: A dict of execution properties.
  
      Returns:
        ExecutionResult proto with anomalies and the component generated alerts
        execution property set with anomalies alerts, if any.
      """
      self._log_startup(input_dict, output_dict, exec_properties)
  
      # Load and deserialize include splits from execution properties.
      include_splits_list = (
          json_utils.loads(
              exec_properties.get(
                  standard_component_specs.INCLUDE_SPLIT_PAIRS_KEY,
                  "null",
              )
          )
          or []
      )
      include_splits = set(
          (test, base) for test, base in include_splits_list
      )
      blessed_value_dict = {}
      executor_output = execution_result_pb2.ExecutorOutput()
  
      anomalies_artifact = artifact_utils.get_single_instance(
          output_dict[standard_component_specs.ANOMALIES_KEY]
      )
  
      test_statistics = artifact_utils.get_single_instance(
          input_dict[standard_component_specs.STATISTICS_KEY]
      )
      if input_dict[standard_component_specs.BASELINE_STATISTICS_KEY]:
          baseline_statistics = artifact_utils.get_single_instance(
              input_dict[
                  standard_component_specs.BASELINE_STATISTICS_KEY
              ]
          )
      else:
          logging.warning(
              "No baseline statistics found. Rubber stamping distribution"
              " validation for all splits."
          )
          for split in artifact_utils.decode_split_names(
              test_statistics.split_names
          ):
              blessed_value_dict[split] = NO_BASELINE_STATS
          anomalies_artifact.set_json_value_custom_property(
              ARTIFACT_PROPERTY_BLESSED_KEY, blessed_value_dict
          )
          executor_output.output_artifacts[
              standard_component_specs.ANOMALIES_KEY
          ].artifacts.append(anomalies_artifact.mlmd_artifact)
          return executor_output
  
      config = exec_properties.get(
          standard_component_specs.DISTRIBUTION_VALIDATOR_CONFIG_KEY
      )
      custom_validation_config = exec_properties.get(
          standard_component_specs.CUSTOM_VALIDATION_CONFIG_KEY
      )
  
      logging.info(
          "Running distribution_validator with config %s", config
      )
  
      # Set up pairs of splits to validate.
      split_pairs = []
      for test_split in artifact_utils.decode_split_names(
          test_statistics.split_names
      ):
          for baseline_split in artifact_utils.decode_split_names(
              baseline_statistics.split_names
          ):
              if (test_split, baseline_split) in include_splits:
                  split_pairs.append((test_split, baseline_split))
              elif not include_splits and test_split == baseline_split:
                  split_pairs.append((test_split, baseline_split))
      if not split_pairs:
          raise ValueError(
              "No split pairs from test and baseline statistics: %s, %s"
              % (test_statistics, baseline_statistics)
          )
      if include_splits:
          missing_split_pairs = include_splits - set(split_pairs)
          if missing_split_pairs:
              raise ValueError(
                  "Missing split pairs identified in include_split_pairs: %s"
                  % ", ".join(
                      [
                          "%s_%s" % (test, baseline)
                          for test, baseline in missing_split_pairs
                      ]
                  )
              )
  
      anomalies_artifact.split_names = (
          artifact_utils.encode_split_names(
              [
                  "%s_%s" % (test, baseline)
                  for test, baseline in split_pairs
              ]
          )
      )
  
      validation_metrics_artifact = None
      if standard_component_specs.VALIDATION_METRICS_KEY in output_dict:
          validation_metrics_artifact = (
              artifact_utils.get_single_instance(
                  output_dict[
                      standard_component_specs.VALIDATION_METRICS_KEY
                  ]
              )
          )
          validation_metrics_artifact.split_names = (
              artifact_utils.encode_split_names(
                  [
                      "%s_%s" % (test, baseline)
                      for test, baseline in split_pairs
                  ]
              )
          )
      current_stats_span = test_statistics.span
      alerts = (
          component_generated_alert_pb2.ComponentGeneratedAlertList()
      )
      for test_split, baseline_split in split_pairs:
          split_pair = "%s_%s" % (test_split, baseline_split)
          logging.info("Processing split pair %s", split_pair)
          test_stats_split = stats_artifact_utils.load_statistics(
              test_statistics, test_split
          ).proto()
          baseline_stats_split = stats_artifact_utils.load_statistics(
              baseline_statistics, baseline_split
          ).proto()
  
          schema = _make_schema_from_config(
              config, baseline_stats_split
          )
          full_anomalies = tfdv.validate_statistics(
              test_stats_split,
              schema,
              previous_statistics=baseline_stats_split,
              custom_validation_config=custom_validation_config,
          )
          anomalies = _get_comparison_only_anomalies(full_anomalies)
          anomalies = _add_anomalies_for_missing_comparisons(
              anomalies, config
          )
  
          if anomalies.anomaly_info or anomalies.HasField(
              "dataset_anomaly_info"
          ):
              blessed_value_dict[split_pair] = NOT_BLESSED_VALUE
          else:
              blessed_value_dict[split_pair] = BLESSED_VALUE
  
          writer_utils.write_anomalies(
              os.path.join(
                  anomalies_artifact.uri,
                  "SplitPair-%s" % split_pair,
                  DEFAULT_FILE_NAME,
              ),
              anomalies,
          )
          monitoring_utils.generate_monitoring_metrics(
              test_stats_split,
              baseline_stats_split,
              split_pair,
              current_stats_span,
              validation_metrics_artifact,
          )
          alerts.component_generated_alert_list.extend(
              _create_anomalies_alerts(anomalies, split_pair)
          )
  
      # Set blessed custom property for Anomalies Artifact
      anomalies_artifact.set_json_value_custom_property(
          ARTIFACT_PROPERTY_BLESSED_KEY, blessed_value_dict
      )
  
      executor_output.output_artifacts[
          standard_component_specs.ANOMALIES_KEY
      ].artifacts.append(anomalies_artifact.mlmd_artifact)
  
      # Set component generated alerts execution property in ExecutorOutput if
      # any anomalies alerts exist.
      if alerts.component_generated_alert_list:
          any_proto = any_pb2.Any()
          any_proto.Pack(alerts)
          executor_output.execution_properties[
              constants.COMPONENT_GENERATED_ALERTS_KEY
          ].proto_value.CopyFrom(any_proto)
  
      return executor_output
  
imports: import os
  from typing import Any, Dict
  from absl import logging
  import tensorflow_data_validation
  from tensorflow_data_validation.utils import path
  from tensorflow_data_validation.utils import schema_util
  from tfx import types
  from tfx.components.statistics_gen import stats_artifact_utils
  from tfx.dsl.components.base import base_executor
  from tfx.orchestration.experimental.core import component_generated_alert_pb2
  from tfx.orchestration.experimental.core import constants
  from tfx.proto import distribution_validator_pb2
  from tfx.proto.orchestration import execution_result_pb2
  from tfx.types import artifact_utils
  from tfx.types import standard_component_specs
  from tfx.utils import json_utils
  from tfx.utils import monitoring_utils
  from tfx.utils import writer_utils
  from google.protobuf import any_pb2
  from tensorflow_metadata.proto.v0 import anomalies_pb2
  from tensorflow_metadata.proto.v0 import schema_pb2
  from tensorflow_metadata.proto.v0 import statistics_pb2
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) and returns: executor_output, executor_output