code: def _to_topk_tuples(
      sliced_record_batch: Tuple[types.SliceKey, pa.RecordBatch],
      bytes_features: FrozenSet[types.FeaturePath],
      categorical_numeric_types: Mapping[
          types.FeaturePath, "schema_pb2.FeatureType"
      ],
      example_weight_map: ExampleWeightMap,
  ) -> Iterable[
      Tuple[
          Tuple[types.SliceKey, types.FeaturePathTuple, Any],
          Tuple[int, Union[int, float]],
      ]
  ]:
      """Generates tuples for computing top-k and uniques from the input."""
      slice_key, record_batch = sliced_record_batch
  
      for (
          feature_path,
          feature_array,
          weights,
      ) in arrow_util.enumerate_arrays(
          record_batch,
          example_weight_map=example_weight_map,
          enumerate_leaves_only=True,
      ):
          feature_array_type = feature_array.type
          feature_type = stats_util.get_feature_type_from_arrow_type(
              feature_path, feature_array_type
          )
          if feature_path in bytes_features:
              continue
          if not _should_run(
              categorical_numeric_types, feature_path, feature_type
          ):
              continue
          flattened_values, parent_indices = array_util.flatten_nested(
              feature_array, weights is not None
          )
          if weights is not None and flattened_values:
              # Slow path: weighted uniques.
              flattened_values_np = np.asarray(flattened_values)
              weights_ndarray = weights[parent_indices]
              for value, count, weight in _weighted_unique(
                  flattened_values_np, weights_ndarray
              ):
                  yield (slice_key, feature_path.steps(), value), (
                      count,
                      weight,
                  )
          else:
              value_counts = flattened_values.value_counts()
              values = value_counts.field("values").to_pylist()
              counts = value_counts.field("counts").to_pylist()
              for value, count in zip(values, counts):
                  yield (
                      (slice_key, feature_path.steps(), value),
                      (count, 1),
                  )
  
imports: import logging
  from typing import Any, FrozenSet, Iterable, Iterator, Mapping, Optional, Text, Tuple, Union
  import apache_beam
  import numpy
  import pandas
  import pyarrow
  from tensorflow_data_validation import types
  from tensorflow_data_validation.arrow import arrow_util
  from tensorflow_data_validation.statistics.generators import stats_generator
  from tensorflow_data_validation.utils import schema_util
  from tensorflow_data_validation.utils import stats_util
  from tensorflow_data_validation.utils import top_k_uniques_stats_util
  from tensorflow_data_validation.utils.example_weight_map import ExampleWeightMap
  from tfx_bsl.arrow import array_util
  from tensorflow_metadata.proto.v0 import schema_pb2
  from tensorflow_metadata.proto.v0 import statistics_pb2
label_desc: Use image node_id 3 for calling a global function with example usage: _to_topk_tuples(sliced_record_batch, bytes_features, categorical_numeric_types, example_weight_map) without return types