Repository: incubator-beam Updated Branches: refs/heads/python-sdk 7c0c27ac0 -> 762a2930a
DoOutputsTuple cleanup This avoids the same (logical) PCollection from being re-created with a different producer. Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/e2eb3b52 Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/e2eb3b52 Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/e2eb3b52 Branch: refs/heads/python-sdk Commit: e2eb3b520ce742bbf4aef63046ed2dfdb96c32e5 Parents: 7c0c27a Author: Robert Bradshaw <[email protected]> Authored: Thu Jul 14 11:03:41 2016 -0700 Committer: Robert Bradshaw <[email protected]> Committed: Thu Jul 14 11:03:41 2016 -0700 ---------------------------------------------------------------------- sdks/python/apache_beam/pvalue.py | 28 ++++++++++++++++------------ 1 file changed, 16 insertions(+), 12 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/e2eb3b52/sdks/python/apache_beam/pvalue.py ---------------------------------------------------------------------- diff --git a/sdks/python/apache_beam/pvalue.py b/sdks/python/apache_beam/pvalue.py index 8552a45..6fc3041 100644 --- a/sdks/python/apache_beam/pvalue.py +++ b/sdks/python/apache_beam/pvalue.py @@ -58,15 +58,15 @@ class PValue(object): self.producer = None def __str__(self): - return '<%s>' % self._str_internal() + return self._str_internal() def __repr__(self): return '<%s at %s>' % (self._str_internal(), hex(id(self))) def _str_internal(self): - return '%s transform=%s' % ( - self.__class__.__name__, - self.producer.transform if self.producer else 'n/a') + return "%s[%s.%s]" % (self.__class__.__name__, + self.producer.full_label if self.producer else None, + self.tag) def apply(self, *args, **kwargs): """Applies a transform or callable to a PValue. @@ -191,16 +191,20 @@ class DoOutputsTuple(object): # Check if we accessed this tag before. if tag in self._pcolls: return self._pcolls[tag] + if tag is not None: self._transform.side_output_tags.add(tag) - pcoll = PCollection(self._pipeline, tag=tag) - # Transfer the producer from the DoOutputsTuple to the resulting - # PCollection. - pcoll.producer = self.producer - # Add this as an output to both the inner ParDo and the outer _MultiParDo - # PTransforms. - self.producer.parts[0].add_output(pcoll, tag) - self.producer.add_output(pcoll, tag) + pcoll = PCollection(self._pipeline, tag=tag) + # Transfer the producer from the DoOutputsTuple to the resulting + # PCollection. + pcoll.producer = self.producer + # Add this as an output to both the inner ParDo and the outer _MultiParDo + # PTransforms. + self.producer.parts[0].add_output(pcoll, tag) + self.producer.add_output(pcoll, tag) + else: + # Main output is output of inner ParDo. + pcoll = self.producer.parts[0].outputs[0] self._pcolls[tag] = pcoll return pcoll
