def _merge_outputs_by_key(keys_and_outputs, outputs_dtype):
  """Merge outputs of analyzers per key into a single output.

  Takes a list of elements of the form (key, [output0, ..., output{N-1}]) and
  returns a list of ndarrays of the form [keys, outputs0, ..., outputs[{N-1}]]
  where keys is formed by stacking the values of `key` from the list and
  similarly outputs{k} is formed by stacking the individual elements of
  output{k} from the list.

  For each k, output{k} must be an ndarray whose size is the same for each
  element of the list.

  Args:
    keys_and_outputs: A list of elements of the form
      (key, [output0, ..., output{N-1}])
    outputs_dtype: A list of tf.DType. Each element corresponds to an output.

  Yields:
    The `TaggedOutput`s: keys, outputs0, ..., outputs[{N-1}]

  Raises:
    ValueError: If the number is outputs doesn't match num_outputs.
  """
  num_outputs = len(outputs_dtype)

  # Sort a copy of keys_and_outputs by keys.
  sorted_keys_and_outputs = sorted(keys_and_outputs, key=lambda x: x[0])

  # Convert from a list of pairs of the form (key, outputs_for_key) to a list of
  # keys and a list of outputs (where the outer dimension is the number of
  # outputs not the number of keys).
  key = []
  outputs = []
  for k, o in sorted_keys_and_outputs:
    key.append(k)
    outputs.append(o)
  if not outputs:
    outputs = [[]] * num_outputs
  else:
    outputs = list(zip(*outputs))
  yield beam.pvalue.TaggedOutput('key',
                                 np.array(key, dtype=tf.string.as_numpy_dtype))
  if len(outputs) != num_outputs:
    raise ValueError(
        'Analyzer has {} outputs but its implementation produced {} '
        'values'.format(num_outputs, len(outputs)))
  for i, (output, dtype) in enumerate(zip(outputs, outputs_dtype)):
    yield beam.pvalue.TaggedOutput(str(i), np.array(output,
                                                    dtype=dtype.as_numpy_dtype))
Merge outputs of analyzers per key into a single output.

  Takes a list of elements of the form (key, [output0, ..., output{N-1}]) and
  returns a list of ndarrays of the form [keys, outputs0, ..., outputs[{N-1}]]
  where keys is formed by stacking the values of `key` from the list and
  similarly outputs{k} is formed by stacking the individual elements of
  output{k} from the list.

  For each k, output{k} must be an ndarray whose size is the same for each
  element of the list.

  Args:
    keys_and_outputs: A list of elements of the form
      (key, [output0, ..., output{N-1}])
    outputs_dtype: A list of tf.DType. Each element corresponds to an output.

  Yields:
    The `TaggedOutput`s: keys, outputs0, ..., outputs[{N-1}]

  Raises:
    ValueError: If the number is outputs doesn't match num_outputs.
  
tensorflow_transform/beam/analyzer_impls.py