[BEAM-3153] Add Python processing-time timers and clock injection #4148

mariapython · 2017-11-18T02:45:52Z

Add processing-time timers
Add clock to runners/direct
Clock injection and propagation:
DirectRunner → EvaluationContext → WatermarkManager
Use RealClock() for production and TestClock() for tests
Add mechanism to detect not-yet fired realtime timers
Refactoring:
_TransformWatermarks’ extract_fired_timers → extract_transform_timers
WatermarkManager’s extract_fired_timers → extract_all_timers
EvaluationContext’s extract_fired_timers → extract_all_timers
Add and improve comments
Tests

asfgit · 2017-11-18T02:46:13Z

FAILURE

--none--

asfgit · 2017-11-18T02:49:03Z

SUCCESS

--none--

asfgit · 2017-11-18T03:11:10Z

FAILURE

--none--

asfgit · 2017-11-18T04:14:15Z

SUCCESS

--none--

mariapython · 2017-11-18T05:07:18Z

R: @charlesccychen

asfgit · 2017-11-19T05:58:10Z

SUCCESS

--none--

asfgit · 2017-11-19T06:50:49Z

SUCCESS

--none--

asfgit · 2017-11-19T08:23:58Z

FAILURE

--none--

mariapython · 2017-11-20T03:03:01Z

retest this please

asfgit · 2017-11-20T03:06:37Z

SUCCESS

--none--

asfgit · 2017-11-20T03:58:46Z

SUCCESS

--none--

asfgit · 2017-11-20T05:33:31Z

FAILURE

--none--

mariapython · 2017-11-21T00:22:45Z

Test timing out, will attempt a 3rd run. Perhaps related to https://issues.apache.org/jira/browse/BEAM-3040

mariapython · 2017-11-21T00:23:04Z

retest this please

charlesccychen

Thanks.

charlesccychen · 2017-11-29T22:19:25Z

sdks/python/apache_beam/runners/direct/direct_runner.py

-  # Currently this only works for overrides where the input and output types do
-  # not change.
-  # For internal SDK use only. This should not be updated by Beam pipeline
-  # authors.


Why the change here?

charlesccychen · 2017-11-29T22:19:25Z

sdks/python/apache_beam/runners/direct/watermark_manager.py

@@ -143,13 +143,18 @@ def _refresh_watermarks(self, applied_ptransform):
            for consumer in consumers:
              self._refresh_watermarks(consumer)

-  def extract_fired_timers(self):
+  def extract_all_timers(self):


Can we use the previous method name? This name is misleading because it isn't we still only extract fired timers.

charlesccychen · 2017-11-29T22:19:26Z

sdks/python/apache_beam/runners/direct/clock.py

-  def time(self):
-    """Returns the number of milliseconds since epoch."""
-    return int(time.time() * 1000)
+  def advance_time(self):


Please add duration argument.

charlesccychen · 2017-11-29T22:19:26Z

sdks/python/apache_beam/runners/direct/clock.py

-  def __init__(self, now_in_ms):
-    self._now_in_ms = now_in_ms
+class TestClock(object):
+  """Clock used for Testing"""


"testing".

Please document the unit used in current_time and advance_by.

charlesccychen · 2017-11-29T22:19:26Z

sdks/python/apache_beam/runners/direct/clock.py

 from __future__ import absolute_import

 import time


 class Clock(object):
-  """For internal use only; no backwards-compatibility guarantees."""
+  def current_time(self):


Can we rename this time() to be consistent with the way time.time() is called?

charlesccychen · 2017-11-29T22:19:26Z

sdks/python/apache_beam/runners/direct/executor.py

-      It will check for successful completion by checking the watermarks of all
-      transforms. If they all reached the maximum watermark it means that
-      pipeline successfully reached to completion.
+      If there is anything in the queue of tasks to do, do not shut down.


Are we checking whether there are pending processing time timers?

charlesccychen · 2017-11-29T22:19:26Z

sdks/python/apache_beam/runners/direct/direct_runner.py

  _PTRANSFORM_OVERRIDES = []

  def __init__(self):
    self._cache = None
+    self._use_test_clock = False  # use RealClock() in production


Can you document that this is used for testing when TestStream is used in the pipeline?

mariapython

PTAL. Thanks!

mariapython · 2017-11-30T21:14:41Z

sdks/python/apache_beam/runners/direct/direct_runner.py

-  # Currently this only works for overrides where the input and output types do
-  # not change.
-  # For internal SDK use only. This should not be updated by Beam pipeline
-  # authors.


charlesccychen wrote:
Why the change here?

At the time of the 2.0 release we tried to normalize the message to "For internal use only; no backwards-compatibility guarantees." (https://issues.apache.org/jira/browse/BEAM-2241)

mariapython · 2017-11-30T21:14:41Z

sdks/python/apache_beam/runners/direct/watermark_manager.py

@@ -143,13 +143,18 @@ def _refresh_watermarks(self, applied_ptransform):
            for consumer in consumers:
              self._refresh_watermarks(consumer)

-  def extract_fired_timers(self):
+  def extract_all_timers(self):


charlesccychen wrote:
Can we use the previous method name? This name is misleading because it isn't we still only extract fired timers.

I meant to do the following changes:
WatermarkManager's extract_fired_timers --> extract_all_timers (meaning for all transforms)
_TransformWatermarks' extract_fired_timers --> extract_transform_timers (meaning per transform).
I meant to accomplish two things: 1) differentiate between "for all" and "per transfer" and 2) we are no longer just extracting the fired timers. I understand 2) is not crucial, as the other return is basically a flag.
Another (verbose) option is: extract_fired_timers and extract_fired_timers_per_transform.

mariapython · 2017-11-30T21:14:42Z

sdks/python/apache_beam/runners/direct/executor.py

-      It will check for successful completion by checking the watermarks of all
-      transforms. If they all reached the maximum watermark it means that
-      pipeline successfully reached to completion.
+      If there is anything in the queue of tasks to do, do not shut down.


charlesccychen wrote:
Are we checking whether there are pending processing time timers?

Yes, and I consider that a "task to do," which is looked at in _is_executing() and that is where I added a comment about timers being set.

mariapython · 2017-11-30T21:14:42Z

sdks/python/apache_beam/runners/direct/clock.py

-  def __init__(self, now_in_ms):
-    self._now_in_ms = now_in_ms
+class TestClock(object):
+  """Clock used for Testing"""


charlesccychen wrote:
"testing".

Please document the unit used in current_time and advance_by.

Done.

mariapython · 2017-11-30T21:14:42Z

sdks/python/apache_beam/runners/direct/clock.py

-  def time(self):
-    """Returns the number of milliseconds since epoch."""
-    return int(time.time() * 1000)
+  def advance_time(self):


charlesccychen wrote:
Please add duration argument.

Done.

mariapython · 2017-11-30T21:14:42Z

sdks/python/apache_beam/runners/direct/clock.py

 from __future__ import absolute_import

 import time


 class Clock(object):
-  """For internal use only; no backwards-compatibility guarantees."""
+  def current_time(self):


charlesccychen wrote:
Can we rename this time() to be consistent with the way time.time() is called?

Done.

mariapython · 2017-11-30T21:14:42Z

sdks/python/apache_beam/runners/direct/direct_runner.py

  _PTRANSFORM_OVERRIDES = []

  def __init__(self):
    self._cache = None
+    self._use_test_clock = False  # use RealClock() in production


charlesccychen wrote:
Can you document that this is used for testing when TestStream is used in the pipeline?

I mention that a few lines below, when I turn on the flag.

mariapython · 2017-12-01T00:50:49Z

retest this please

aaltay · 2017-12-06T07:53:16Z

What is the next action on this PR? @charlesccychen could you please make another pass?

charlesccychen

Thanks.

charlesccychen · 2017-12-09T00:24:16Z

sdks/python/apache_beam/runners/direct/clock.py

-  """For internal use only; no backwards-compatibility guarantees.
+class RealClock(object):
+  def time(self):
+    return int(time.time() * 1000)


Can you please change this back to seconds?

charlesccychen · 2017-12-09T00:24:17Z

sdks/python/apache_beam/transforms/trigger.py

@@ -1138,20 +1138,36 @@ def clear_state(self, window, tag):
    if not self.state[window]:
      self.state.pop(window, None)

-  def get_timers(self, clear=False, watermark=MAX_TIMESTAMP):
+  def get_timers(self, clear=False, watermark=MAX_TIMESTAMP, current_time=None):


Let's change this argument name to "processing_time".

charlesccychen · 2017-12-09T00:24:17Z

sdks/python/apache_beam/transforms/trigger.py

    for window, timers in list(self.timers.items()):
      for (name, time_domain), timestamp in list(timers.items()):
-        if timestamp <= watermark:
+        if time_domain == 'REAL_TIME':


Please use the TimeDomain enum (i.e. TimeDomain.WATERMARK, etc) here.

charlesccychen · 2017-12-09T00:24:17Z

sdks/python/apache_beam/transforms/trigger.py

@@ -1138,20 +1138,36 @@ def clear_state(self, window, tag):
    if not self.state[window]:
      self.state.pop(window, None)

-  def get_timers(self, clear=False, watermark=MAX_TIMESTAMP):
+  def get_timers(self, clear=False, watermark=MAX_TIMESTAMP, current_time=None):
+    """Gets expired timers and reports if there


Let's try to fit in one line:

"""Returns expired timers and existence of REAL_TIME timers."""

mariapython · 2017-12-10T20:11:46Z

retest this please

mariapython

Thanks! PTAL

mariapython · 2017-12-11T10:07:15Z

sdks/python/apache_beam/runners/direct/clock.py

-  """For internal use only; no backwards-compatibility guarantees.
+class RealClock(object):
+  def time(self):
+    return int(time.time() * 1000)


charlesccychen wrote:
Can you please change this back to seconds?

Done.

mariapython · 2017-12-11T10:07:16Z

sdks/python/apache_beam/transforms/trigger.py

@@ -1138,20 +1138,36 @@ def clear_state(self, window, tag):
    if not self.state[window]:
      self.state.pop(window, None)

-  def get_timers(self, clear=False, watermark=MAX_TIMESTAMP):
+  def get_timers(self, clear=False, watermark=MAX_TIMESTAMP, current_time=None):


charlesccychen wrote:
Let's change this argument name to "processing_time".

Done.

mariapython · 2017-12-11T10:07:16Z

sdks/python/apache_beam/transforms/trigger.py

    for window, timers in list(self.timers.items()):
      for (name, time_domain), timestamp in list(timers.items()):
-        if timestamp <= watermark:
+        if time_domain == 'REAL_TIME':


charlesccychen wrote:
Please use the TimeDomain enum (i.e. TimeDomain.WATERMARK, etc) here.

Done.

mariapython · 2017-12-11T10:07:16Z

sdks/python/apache_beam/transforms/trigger.py

@@ -1138,20 +1138,36 @@ def clear_state(self, window, tag):
    if not self.state[window]:
      self.state.pop(window, None)

-  def get_timers(self, clear=False, watermark=MAX_TIMESTAMP):
+  def get_timers(self, clear=False, watermark=MAX_TIMESTAMP, current_time=None):
+    """Gets expired timers and reports if there


charlesccychen wrote:
Let's try to fit in one line:

"""Returns expired timers and existence of REAL_TIME timers."""

Yes, it can be done for the first sentence, but I think the second is more important (explaining the addition of 'time_marker'). When I added the docstring initially, I considered writing that second sentence by the 'if' but I found it more clunky and, since the expiration is mentioned in the first sentence, referring to it in the second flowed better.

charlesccychen

Thanks.

mariapython wrote:
Thanks! PTAL

Done.

charlesccychen · 2017-12-14T21:34:51Z

sdks/python/apache_beam/runners/direct/clock.py

+    raise NotImplementedError()
+
+  def advance_time(self, advance_by):
+    """Advances the clock by a number of miliseconds."""


Please fix comment to indicate seconds.

charlesccychen · 2017-12-14T21:34:51Z

sdks/python/apache_beam/runners/direct/clock.py

 from __future__ import absolute_import

 import time


 class Clock(object):
-  """For internal use only; no backwards-compatibility guarantees."""
-
  def time(self):
    """Returns the number of milliseconds since epoch."""


Please fix comment to indicate seconds.

charlesccychen · 2017-12-14T21:34:51Z

sdks/python/apache_beam/runners/direct/executor.py

+
+      watermark_manager = self._executor.evaluation_context._watermark_manager
+      _, any_unfired_timers = watermark_manager.extract_all_timers()
+      if any_unfired_timers:


Can you more clearly indicate (by naming or comment) that this is for unfired real time timers?

mariapython

Thanks. PTAL.

mariapython · 2017-12-14T22:19:55Z

sdks/python/apache_beam/runners/direct/clock.py

+    raise NotImplementedError()
+
+  def advance_time(self, advance_by):
+    """Advances the clock by a number of miliseconds."""


charlesccychen wrote:
Please fix comment to indicate seconds.

Done.

mariapython · 2017-12-14T22:19:56Z

sdks/python/apache_beam/runners/direct/clock.py

 from __future__ import absolute_import

 import time


 class Clock(object):
-  """For internal use only; no backwards-compatibility guarantees."""
-
  def time(self):
    """Returns the number of milliseconds since epoch."""


charlesccychen wrote:
Please fix comment to indicate seconds.

Done.

mariapython · 2017-12-14T22:19:56Z

sdks/python/apache_beam/runners/direct/executor.py

+
+      watermark_manager = self._executor.evaluation_context._watermark_manager
+      _, any_unfired_timers = watermark_manager.extract_all_timers()
+      if any_unfired_timers:


charlesccychen wrote:
Can you more clearly indicate (by naming or comment) that this is for unfired real time timers?

Done.

charlesccychen

Thanks, LGTM.

mariapython wrote:
Thanks. PTAL.

Done.

charlesccychen

mariapython wrote:
PTAL. Thanks!

Done.

chamikaramj · 2017-12-15T01:41:23Z

Thanks. Merging.

charlesccychen reviewed Nov 29, 2017

View reviewed changes

mariapython force-pushed the direct-timers-show branch from 2ab61a1 to 0f103d9 Compare November 30, 2017 19:12

mariapython commented Nov 30, 2017

View reviewed changes

charlesccychen reviewed Dec 9, 2017

View reviewed changes

mariapython force-pushed the direct-timers-show branch from 0f103d9 to c1b2152 Compare December 9, 2017 22:10

mariapython commented Dec 11, 2017

View reviewed changes

charlesccychen reviewed Dec 14, 2017

View reviewed changes

Add Python processing-time timers and clock injection

6bcf84f

mariapython added 5 commits December 14, 2017 14:13

Fix linter error

ec7615b

Add review comments

516e04c

Address review comments

679a92d

borrar

d1db15d

Add review commnets

79e167f

mariapython force-pushed the direct-timers-show branch from d675132 to 79e167f Compare December 14, 2017 22:14

mariapython commented Dec 14, 2017

View reviewed changes

charlesccychen approved these changes Dec 14, 2017

View reviewed changes

chamikaramj merged commit 734b2be into apache:master Dec 15, 2017

This pull request was closed.

[BEAM-3153] Add Python processing-time timers and clock injection #4148

[BEAM-3153] Add Python processing-time timers and clock injection #4148

Conversation

mariapython commented Nov 18, 2017

asfgit commented Nov 18, 2017

asfgit commented Nov 18, 2017

asfgit commented Nov 18, 2017

asfgit commented Nov 18, 2017

mariapython commented Nov 18, 2017

asfgit commented Nov 19, 2017

asfgit commented Nov 19, 2017

asfgit commented Nov 19, 2017

mariapython commented Nov 20, 2017

asfgit commented Nov 20, 2017

asfgit commented Nov 20, 2017

asfgit commented Nov 20, 2017

mariapython commented Nov 21, 2017

mariapython commented Nov 21, 2017

charlesccychen 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

mariapython 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

mariapython commented Dec 1, 2017

aaltay commented Dec 6, 2017

charlesccychen 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

mariapython commented Dec 10, 2017

mariapython 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

charlesccychen 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

mariapython 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

charlesccychen left a comment

Choose a reason for hiding this comment

charlesccychen left a comment

Choose a reason for hiding this comment

chamikaramj commented Dec 15, 2017