[BEAM-1630] Adds API for defining Splittable DoFns using Python SDK. #3882

chamikaramj · 2017-09-22T00:45:48Z

See https://s.apache.org/splittable-do-fn-python-sdk for the design.

This PR and the above doc were updated to reflect following recent updates to Splittable DoFn.

Support for ProcessContinuations
Support for dynamically updating output watermark irrespective of the output element production.

This will be followed by a PR that adds support for reading Splittable DoFns using DirectRunner.

See https://s.apache.org/splittable-do-fn-python-sdk for the design. This PR and the above doc were updated to reflect following recent updates to Splittable DoFn. * Support for ProcessContinuations * Support for dynamically updating output watermark irrespective of the output element production. This will be followed by a PR that adds support for reading Splittable DoFns using DirectRunner.

chamikaramj · 2017-09-22T00:46:15Z

R: @robertwb

cc: @jkff

jkff

Thanks!

jkff · 2017-09-22T00:59:29Z

sdks/python/apache_beam/transforms/core.py

+    need to write a Splittable DoFn.
+
+    Not all runners support Splittable DoFn. See the capability matrix
+    (a href="https://beam.apache.org/documentation/runners/capability-matrix/)


The matrix is currently focused on Java, so this is a bit misleading.

Removed this paragraph.

jkff · 2017-09-22T01:00:13Z

sdks/python/apache_beam/transforms/core.py

+    A Splittable DoFn must provide suitable overrides for the following methods
+    of the ``DoFn`` class.
+    * new_tracker()
+    * restriction_coder()


In Java, restriction_coder() and split() are not required (have defaults). Will this be addressed in Python too?

Made those two methods optional and added default implementations.

jkff · 2017-09-22T01:01:45Z

sdks/python/apache_beam/transforms/core.py

+    pass
+
+  @staticmethod
+  def stop():


Is this necessary in Python? Seems like with the "last yielded element is a ProcessContinuation" thing, instead of saying yield stop() you can simply not yield anything.

Yeah, makes sense to remove this. Removed and updated docs.

chamikaramj

Thanks.

chamikaramj · 2017-09-22T04:12:05Z

sdks/python/apache_beam/transforms/core.py

+    need to write a Splittable DoFn.
+
+    Not all runners support Splittable DoFn. See the capability matrix
+    (a href="https://beam.apache.org/documentation/runners/capability-matrix/)


Removed this paragraph.

chamikaramj · 2017-09-22T04:12:37Z

sdks/python/apache_beam/transforms/core.py

+    A Splittable DoFn must provide suitable overrides for the following methods
+    of the ``DoFn`` class.
+    * new_tracker()
+    * restriction_coder()


Made those two methods optional and added default implementations.

chamikaramj · 2017-09-22T04:13:02Z

sdks/python/apache_beam/transforms/core.py

+    pass
+
+  @staticmethod
+  def stop():


Yeah, makes sense to remove this. Removed and updated docs.

coveralls · 2017-09-22T05:36:00Z

Coverage increased (+0.03%) to 69.61% when pulling 95a209d on chamikaramj:sdf_api into d6c6339 on apache:master.

jkff

Thanks, delegating the rest of the review to Robert.

chamikaramj · 2017-09-27T01:10:57Z

Friendly ping :)

robertwb · 2017-09-27T17:29:09Z

sdks/python/apache_beam/io/iobase.py

+  See following documents for more details.
+  * https://s.apache.org/splittable-do-fn
+  * https://s.apache.org/splittable-do-fn-python-sdk
+  """


Please document which methods maybe called concurrently (and hence need locking).

robertwb · 2017-09-27T17:30:18Z

sdks/python/apache_beam/io/iobase.py

+    """Returns the current restriction.
+
+    Returns a restriction accurately describing the full range of work the
+    current ``DoFn.process()`` call will do, including already completed work.


Will this be updated over time? Should we note that anything in this restriction is a candidate for removal, and try_claim must be called before actually doing any work?

Currently we don't have try_claim() (and other methods needed for dynamic work rebalancing). For now I mentioned that current restriction might be updated dynamically.

...due to concurrent invocation of other methods...

I think the fact that this may be called from other threads is key.

robertwb · 2017-09-27T17:36:00Z

sdks/python/apache_beam/io/iobase.py

+    fully read.
+
+    Returns: ``True`` if current restriction has been fully processed.
+    Raises ValueError: if there is still any unclaimed work remaining in the


Isn't it legal for a DoFn to return before fully processing the restriction?

Seems like it'd be better to call checkpoint() after the process has completed, at which point, if the entire restriction was processed, the current (remaining) restriction would be None.

Yeah, the runner will take a checkpoint after the processing has completed. This method will be called after that as an additional check to make sure that there are no data remaining in the current restriction after checkpoint is taken.

robertwb · 2017-09-27T17:36:32Z

sdks/python/apache_beam/io/iobase.py

+    """
+    raise NotImplementedError
+
+  def checkpoint(self):


What about try_split? I'd like to avoid redundancy with that one.

Currently we don't support dynamic work rebalancing (neither does Java SDF). I think we should have further discussions before finalizing the API needed for that.

We shouldn't introduce a method that we know will be redundant with future developments.

robertwb · 2017-09-27T17:38:53Z

sdks/python/apache_beam/transforms/core.py

+    single parameter of type ``Timestamp`` or as an integer that gives the
+    watermark in number of seconds.
+
+    ** Splittable DoFns **


Perhaps move the Args section up to here (or maybe even higher). It does feel odd that 99% of DoFns won't be splittable, but 80% of this docstring is about splittable DoFns.

Good point, done :).

robertwb · 2017-09-27T17:40:36Z

sdks/python/apache_beam/transforms/core.py

+    yield restriction
+
+  @staticmethod
+  def resume(resume_delay=0):


Is this intended to be used as

return self.resume(...)

Maybe it'd be better to put this as a static method on ProcessContinuation.

robertwb · 2017-09-27T17:44:19Z

sdks/python/apache_beam/transforms/core.py

@@ -189,6 +276,54 @@ def finish_bundle(self):
    """
    pass

+  def new_tracker(self, restriction):


It's unclear that all of these methods are only used in the splitable case. Maybe some documentation? (Maybe the names should all reflect this, e.g. containing the word "splittable"?)

I might go so far as to put these in a mixin class (interface) that would be checked for if the splittable argument was used, which we could also then to enforce that they're implemented.

Updated documentation. I don't think it makes sense to move these to a MixIn since we don't expect any type other than DoFn to have this functionality.

The advantage is that we can make these many methods abstract such that the user won't forget any of them when they need them.

chamikaramj

Thanks. PTAL.

chamikaramj · 2017-09-28T00:55:13Z

sdks/python/apache_beam/io/iobase.py

+  See following documents for more details.
+  * https://s.apache.org/splittable-do-fn
+  * https://s.apache.org/splittable-do-fn-python-sdk
+  """


chamikaramj · 2017-09-28T00:55:53Z

sdks/python/apache_beam/io/iobase.py

+    """Returns the current restriction.
+
+    Returns a restriction accurately describing the full range of work the
+    current ``DoFn.process()`` call will do, including already completed work.


Currently we don't have try_claim() (and other methods needed for dynamic work rebalancing). For now I mentioned that current restriction might be updated dynamically.

chamikaramj · 2017-09-28T01:01:14Z

sdks/python/apache_beam/io/iobase.py

+    """
+    raise NotImplementedError
+
+  def checkpoint(self):


Currently we don't support dynamic work rebalancing (neither does Java SDF). I think we should have further discussions before finalizing the API needed for that.

chamikaramj · 2017-09-28T01:06:39Z

sdks/python/apache_beam/transforms/core.py

+    single parameter of type ``Timestamp`` or as an integer that gives the
+    watermark in number of seconds.
+
+    ** Splittable DoFns **


Good point, done :).

chamikaramj · 2017-09-28T01:10:24Z

sdks/python/apache_beam/transforms/core.py

+    yield restriction
+
+  @staticmethod
+  def resume(resume_delay=0):


chamikaramj · 2017-09-28T01:14:05Z

sdks/python/apache_beam/transforms/core.py

@@ -189,6 +276,54 @@ def finish_bundle(self):
    """
    pass

+  def new_tracker(self, restriction):


Updated documentation. I don't think it makes sense to move these to a MixIn since we don't expect any type other than DoFn to have this functionality.

chamikaramj · 2017-09-28T02:31:38Z

sdks/python/apache_beam/io/iobase.py

+    fully read.
+
+    Returns: ``True`` if current restriction has been fully processed.
+    Raises ValueError: if there is still any unclaimed work remaining in the


Yeah, the runner will take a checkpoint after the processing has completed. This method will be called after that as an additional check to make sure that there are no data remaining in the current restriction after checkpoint is taken.

chamikaramj · 2017-09-28T23:43:46Z

Failure seems to be unrelated.

Retest this please.

coveralls · 2017-09-29T01:03:11Z

Coverage decreased (-0.01%) to 69.561% when pulling 10067f0 on chamikaramj:sdf_api into d6c6339 on apache:master.

robertwb

Sorry, I didn't realize I handn't sent these comments out yet.

robertwb · 2017-09-27T19:00:13Z

sdks/python/apache_beam/transforms/core.py

+  """
+
+  def __init__(self, should_resume, resume_delay=0):
+    """Initializes a ProcessContinuation object.


Do we need the "should_resume" parameter if it's never returned (yielded) from a DoFn that shouldn't?

robertwb · 2017-09-28T16:56:39Z

sdks/python/apache_beam/io/iobase.py

+    """
+    raise NotImplementedError
+
+  def checkpoint(self):


We shouldn't introduce a method that we know will be redundant with future developments.

robertwb · 2017-09-28T16:58:07Z

sdks/python/apache_beam/transforms/core.py

@@ -189,6 +276,54 @@ def finish_bundle(self):
    """
    pass

+  def new_tracker(self, restriction):


The advantage is that we can make these many methods abstract such that the user won't forget any of them when they need them.

chamikaramj

Thanks.

chamikaramj · 2017-10-02T14:49:27Z

sdks/python/apache_beam/io/iobase.py

+    """
+    raise NotImplementedError
+
+  def checkpoint(self):


robertwb wrote:
We shouldn't introduce a method that we know will be redundant with future developments.

Seems like, based on previous discussions [1], we are moving towards maintaining both methods (try_split() and checkpoint()) due to the need for calling checkpoint() after a resume() ? Also Java API currently has the checkpoint() method [2], we should add this to Python for consistency and if needed update both APIs (again maintaining consistency) after future discussions/experiments ?

[1] https://docs.google.com/document/d/1BGc8pM1GOvZhwR9SARSVte-20XEoBUxrGJ5gTWXdv3c/edit
[2] https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/RestrictionTracker.java#L42

Semantically, checkpoint() enforces that immediately following claim (try_claim() once we have that method) should be rejected while try_split(0) will not enforce that.

chamikaramj · 2017-10-02T14:49:28Z

sdks/python/apache_beam/transforms/core.py

+  """
+
+  def __init__(self, should_resume, resume_delay=0):
+    """Initializes a ProcessContinuation object.


robertwb wrote:
Do we need the "should_resume" parameter if it's never returned (yielded) from a DoFn that shouldn't?

Removed.

chamikaramj · 2017-10-02T14:49:28Z

sdks/python/apache_beam/transforms/core.py

@@ -189,6 +276,54 @@ def finish_bundle(self):
    """
    pass

+  def new_tracker(self, restriction):


robertwb wrote:
The advantage is that we can make these many methods abstract such that the user won't forget any of them when they need them.

I'm not 100% sure of this one. Seems like you are suggesting to ask users to directly implement the MixIn (say SplittableMixIn). But this means that every SDF author will have to implement two classes due to the need to make it easy for the SDK to perform validation. I think it might be better to make the API easy for users (just implement DoFn and implement required SDF method if needed) and make validation the hard (slightly) way.

coveralls · 2017-10-02T17:14:36Z

Coverage decreased (-0.008%) to 69.568% when pulling d538a57 on chamikaramj:sdf_api into d6c6339 on apache:master.

robertwb · 2017-10-04T00:20:53Z

To follow up, I'm OK with leaving checkpoint() for parity with Java until we add try_split. Of note, the former can be called concurrently with DoFn.process, but also is called whenever resume is returned and in this case requires stricter semantics.

Regarding the mixin/documentation issues, here's a better idea: Rather than take a contentless DoFn.RestrictionTrackerParam identifier, have the RestrictionTracker parameter have a default argument that would be a RestrictionProvider instance. This new type would hold all the methods for dealing with restrictions (element -> initial restriction, restriction -> restriction* splitting, and restriction -> tracker, and restriction coder) and of course be a natural place to put all the documentation as well.

chamikaramj · 2017-10-04T03:20:07Z

Thanks.

Added 'RestrictionProvider' and updated documentation accordingly.

PTAL.

chamikaramj · 2017-10-04T03:20:27Z

Retest this please.

robertwb

Looking good!

robertwb · 2017-10-04T22:47:28Z

sdks/python/apache_beam/io/iobase.py

+    """Returns the current restriction.
+
+    Returns a restriction accurately describing the full range of work the
+    current ``DoFn.process()`` call will do, including already completed work.


...due to concurrent invocation of other methods...

I think the fact that this may be called from other threads is key.

robertwb · 2017-10-04T22:52:03Z

sdks/python/apache_beam/transforms/core.py

+  Splittable ``DoFn``s.
+
+  To denote a ``DoFn`` class to be Splittable ``DoFn``, ``DoFn.process()``
+  method of that class should have exactly one parameter of a type that is a


one parameter whose default value is an instance of RestrictionProvider.

robertwb · 2017-10-04T22:55:55Z

sdks/python/apache_beam/transforms/core.py

+  gives the watermark in number of seconds.
+  """
+
+  def new_tracker(self, restriction):


s/new/create/ (which is consistent with the rest of our API).

robertwb · 2017-10-04T22:58:50Z

sdks/python/apache_beam/transforms/core.py

+    """
+    raise NotImplementedError
+
+  def restriction_coder(self):


Move this one last.

robertwb · 2017-10-04T23:07:27Z

sdks/python/apache_beam/transforms/core.py

+    ``DoFn.SideInputParam``: a side input that may be used when processing.
+    ``DoFn.TimestampParam``: timestamp of the input element.
+    ``DoFn.WindowParam``: ``Window`` the input element belongs to.
+    An object of type ``RestrictionProvider``: having a parameter of a type that


An iobase.RestrictionProvider instance: a restriction tracker will be provided here to allow treatment as a Splittable `DoFn``.

chamikaramj

Thanks. PTAL.

chamikaramj · 2017-10-05T05:23:34Z

sdks/python/apache_beam/io/iobase.py

+    """Returns the current restriction.
+
+    Returns a restriction accurately describing the full range of work the
+    current ``DoFn.process()`` call will do, including already completed work.


chamikaramj · 2017-10-05T05:25:15Z

sdks/python/apache_beam/transforms/core.py

+  Splittable ``DoFn``s.
+
+  To denote a ``DoFn`` class to be Splittable ``DoFn``, ``DoFn.process()``
+  method of that class should have exactly one parameter of a type that is a


chamikaramj · 2017-10-05T05:26:05Z

sdks/python/apache_beam/transforms/core.py

+  gives the watermark in number of seconds.
+  """
+
+  def new_tracker(self, restriction):


chamikaramj · 2017-10-05T05:26:53Z

sdks/python/apache_beam/transforms/core.py

+    """
+    raise NotImplementedError
+
+  def restriction_coder(self):


chamikaramj · 2017-10-05T05:27:51Z

sdks/python/apache_beam/transforms/core.py

+    ``DoFn.SideInputParam``: a side input that may be used when processing.
+    ``DoFn.TimestampParam``: timestamp of the input element.
+    ``DoFn.WindowParam``: ``Window`` the input element belongs to.
+    An object of type ``RestrictionProvider``: having a parameter of a type that


chamikaramj · 2017-10-05T16:55:24Z

Retest this please

coveralls · 2017-10-05T19:07:02Z

Coverage increased (+0.003%) to 69.579% when pulling f05b4cc on chamikaramj:sdf_api into d6c6339 on apache:master.

chamikaramj · 2017-10-05T21:14:49Z

Jenkins failure above is unrelated.

Retest this please.

robertwb

LGTM, thanks!

robertwb · 2017-10-05T21:33:25Z

sdks/python/apache_beam/io/iobase.py

+    ``current_restriction()`` and the return value of this method invocation
+    combined.
+
+    This method must be called at most once on a given ``RestrictionTracker``


Nit: we can't enforce or require this, as we may call checkpint immediately before a resume is returned.

chamikaramj · 2017-10-05T22:01:23Z

Thanks.

coveralls · 2017-10-05T23:00:48Z

Coverage increased (+0.003%) to 69.579% when pulling cad1178 on chamikaramj:sdf_api into d6c6339 on apache:master.

coveralls · 2017-10-05T23:29:16Z

Coverage increased (+0.006%) to 69.582% when pulling cad1178 on chamikaramj:sdf_api into d6c6339 on apache:master.

jkff requested changes Sep 22, 2017

View reviewed changes

Addressing reviewer comments.

95a209d

chamikaramj commented Sep 22, 2017

View reviewed changes

jkff reviewed Sep 22, 2017

View reviewed changes

robertwb reviewed Sep 27, 2017

View reviewed changes

Addresses reviewer comments.

10067f0

chamikaramj commented Sep 28, 2017

View reviewed changes

robertwb reviewed Sep 29, 2017

View reviewed changes

Removed should_resume.

d538a57

chamikaramj commented Oct 2, 2017

View reviewed changes

Added class RestrictionProvider and documentation updates.

3816760

robertwb reviewed Oct 4, 2017

View reviewed changes

Addresses reviewer comments.

f05b4cc

chamikaramj commented Oct 5, 2017

View reviewed changes

robertwb reviewed Oct 5, 2017

View reviewed changes

Addresses reviewer comments.

cad1178

asfgit closed this in 31da49c Oct 6, 2017

[BEAM-1630] Adds API for defining Splittable DoFns using Python SDK. #3882

[BEAM-1630] Adds API for defining Splittable DoFns using Python SDK. #3882

Conversation

chamikaramj commented Sep 22, 2017

chamikaramj commented Sep 22, 2017

jkff left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chamikaramj left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

coveralls commented Sep 22, 2017

jkff left a comment

Choose a reason for hiding this comment

chamikaramj commented Sep 27, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chamikaramj left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chamikaramj commented Sep 28, 2017

coveralls commented Sep 29, 2017

robertwb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chamikaramj left a comment

Choose a reason for hiding this comment

chamikaramj Oct 2, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chamikaramj Oct 2, 2017 • edited

Choose a reason for hiding this comment

coveralls commented Oct 2, 2017

robertwb commented Oct 4, 2017

chamikaramj commented Oct 4, 2017

chamikaramj commented Oct 4, 2017

robertwb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chamikaramj left a comment

chamikaramj Oct 2, 2017 •

edited

chamikaramj Oct 2, 2017 •

edited