def deep_copy(pcollection):
  """Create a deep copy of a PCollection up to materialization boundaries."""
  if not isinstance(pcollection, pvalue.PCollection):
    raise ValueError('Input to deep_copy must be a PCollection.')

  # AppliedPTransform.update_input_refcounts() is a vestigial method that
  # uses an incorrect heuristic; it will be removed in a future version of
  # Beam, since its results aren't used anyway.  Until then, we work around
  # this (see https://issues.apache.org/jira/browse/BEAM-4593).
  if getattr(beam_pipeline.AppliedPTransform,
             'update_input_refcounts', None) is not None:
    beam_pipeline.AppliedPTransform.update_input_refcounts = lambda _: None

  to_clone = _get_items_to_clone(pcollection)
  pcollection_replacements = _clone_items(pcollection.pipeline, to_clone)

  return pcollection_replacements[pcollection]
Create a deep copy of a PCollection up to materialization boundaries.
tensorflow_transform/beam/deep_copy.py