[BEAM-2718] Improve performance of bundle retry by mariapython · Pull Request #3815 · apache/beam

mariapython · 2017-09-07T04:21:17Z

In order to keep a consistent state while handling bundles, every time the state is accessed, a copy is made and kept until the bundle is committed. Here a custom copy is used replicating only the inner structures that are needed and shallow-copying the rest.

coveralls · 2017-09-07T05:54:47Z

Coverage increased (+0.007%) to 69.691% when pulling eba397b on mariapython:retry_custom into 6dfd3a2 on apache:master.

coveralls · 2017-09-07T08:09:33Z

Coverage increased (+0.003%) to 69.693% when pulling 8650ea6 on mariapython:retry_custom into df1476d on apache:master.

mariapython · 2017-09-07T18:44:00Z

R: @charlesccychen

charlesccychen

Thanks!

charlesccychen · 2017-09-08T14:57:37Z

sdks/python/apache_beam/transforms/trigger.py

+    cloned_object.global_state = copy.deepcopy(self.global_state)
+    cloned_object.state = copy.copy(self.state)
+    for window in self.state:
+      cloned_object.state[window] = copy.copy(self.state[window])


This can just be an empty dictionary.

charlesccychen · 2017-09-08T14:57:37Z

sdks/python/apache_beam/transforms/trigger.py

+  def custom_copy(self):
+    cloned_object = copy.copy(self)
+    cloned_object.timers = copy.deepcopy(self.timers)
+    cloned_object.global_state = copy.deepcopy(self.global_state)


We should do similar logic to the inner loop below here too.

charlesccychen · 2017-09-08T14:57:38Z

sdks/python/apache_beam/transforms/trigger.py

+    cloned_object = copy.copy(self)
+    cloned_object.timers = copy.deepcopy(self.timers)
+    cloned_object.global_state = copy.deepcopy(self.global_state)
+    cloned_object.state = copy.copy(self.state)


We can just make this an empty dictionary.

charlesccychen · 2017-09-08T14:57:38Z

sdks/python/apache_beam/transforms/trigger.py

    self.global_state = {}
    self.defensive_copy = defensive_copy

+  def custom_copy(self):


Let's just rename this to clone().

charlesccychen · 2017-09-08T14:57:38Z

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

    if not self.partial_keyed_state.get(key):
-      self.partial_keyed_state[key] = self.existing_keyed_state[key].clone()
+      self.partial_keyed_state[key] = (
+          self.existing_keyed_state[key].custom_copy())


Let's just rename this to clone().

mariapython

PTAL

mariapython · 2017-09-13T07:10:20Z

sdks/python/apache_beam/transforms/trigger.py

+    cloned_object = copy.copy(self)
+    cloned_object.timers = copy.deepcopy(self.timers)
+    cloned_object.global_state = copy.deepcopy(self.global_state)
+    cloned_object.state = copy.copy(self.state)


charlesccychen wrote:
We can just make this an empty dictionary.

Done.

mariapython · 2017-09-13T07:10:20Z

sdks/python/apache_beam/transforms/trigger.py

    self.global_state = {}
    self.defensive_copy = defensive_copy

+  def custom_copy(self):


charlesccychen wrote:
Let's just rename this to clone().

See my comments above.

mariapython · 2017-09-13T07:10:20Z

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

    if not self.partial_keyed_state.get(key):
-      self.partial_keyed_state[key] = self.existing_keyed_state[key].clone()
+      self.partial_keyed_state[key] = (
+          self.existing_keyed_state[key].custom_copy())


charlesccychen wrote:
Let's just rename this to clone().

I preferred to avoid that name, as it was not really a deep copy. I changed it to copy().

mariapython · 2017-09-13T07:10:20Z

sdks/python/apache_beam/transforms/trigger.py

+    cloned_object.global_state = copy.deepcopy(self.global_state)
+    cloned_object.state = copy.copy(self.state)
+    for window in self.state:
+      cloned_object.state[window] = copy.copy(self.state[window])


charlesccychen wrote:
This can just be an empty dictionary.

Done.

mariapython · 2017-09-13T07:10:20Z

sdks/python/apache_beam/transforms/trigger.py

+  def custom_copy(self):
+    cloned_object = copy.copy(self)
+    cloned_object.timers = copy.deepcopy(self.timers)
+    cloned_object.global_state = copy.deepcopy(self.global_state)


charlesccychen wrote:
We should do similar logic to the inner loop below here too.

global_state doesn't appear to be used by the transform evaluators. In fact, I was thinking of eliminating the line completely (same reason why defensive_copy is not copied).

mariapython

mariapython wrote:
PTAL

Done.

coveralls · 2017-09-14T01:02:33Z

Coverage decreased (-0.01%) to 69.493% when pulling 3f1a714 on mariapython:retry_custom into 50532f0 on apache:master.

charlesccychen

Thanks!

charlesccychen · 2017-09-14T20:17:42Z

sdks/python/apache_beam/transforms/trigger.py

+    cloned_object.state = (
+        collections.defaultdict(lambda: collections.defaultdict(list)))
+    for window in self.state:
+      cloned_object.state[window] = collections.defaultdict(list)


You can omit this line--the dict is created for you by the defaultdict mechanism on cloned_object.state.

charlesccychen · 2017-09-14T20:17:43Z

sdks/python/apache_beam/transforms/trigger.py

+    cloned_object.timers = copy.deepcopy(self.timers)
+    cloned_object.global_state = copy.deepcopy(self.global_state)
+    cloned_object.state = (
+        collections.defaultdict(lambda: collections.defaultdict(list)))


You can remove these two lines if you use the approach in the comment above.

charlesccychen · 2017-09-14T20:17:43Z

sdks/python/apache_beam/transforms/trigger.py

    self.defensive_copy = defensive_copy

+  def copy(self):
+    cloned_object = copy.copy(self)


Let's omit the copy and just create a new object:

cloned_object = InMemoryUnmergedState(defensive_copy=self.defensive_copy)

mariapython · 2017-09-15T01:49:40Z

sdks/python/apache_beam/transforms/trigger.py

+    cloned_object.timers = copy.deepcopy(self.timers)
+    cloned_object.global_state = copy.deepcopy(self.global_state)
+    cloned_object.state = (
+        collections.defaultdict(lambda: collections.defaultdict(list)))


charlesccychen wrote:
You can remove these two lines if you use the approach in the comment above.

Done.

mariapython · 2017-09-15T01:49:40Z

sdks/python/apache_beam/transforms/trigger.py

+    cloned_object.state = (
+        collections.defaultdict(lambda: collections.defaultdict(list)))
+    for window in self.state:
+      cloned_object.state[window] = collections.defaultdict(list)


charlesccychen wrote:
You can omit this line--the dict is created for you by the defaultdict mechanism on cloned_object.state.

Done.

mariapython · 2017-09-15T01:49:40Z

sdks/python/apache_beam/transforms/trigger.py

    self.defensive_copy = defensive_copy

+  def copy(self):
+    cloned_object = copy.copy(self)


charlesccychen wrote:
Let's omit the copy and just create a new object:

cloned_object = InMemoryUnmergedState(defensive_copy=self.defensive_copy)

Done.

coveralls · 2017-09-15T02:53:51Z

Coverage decreased (-0.01%) to 69.496% when pulling 585d8f9 on mariapython:retry_custom into 307b036 on apache:master.

charlesccychen

Thanks!

R: @chamikaramj for merge

mariapython force-pushed the retry_custom branch from eba397b to 8650ea6 Compare September 7, 2017 06:34

charlesccychen reviewed Sep 8, 2017

View reviewed changes

mariapython force-pushed the retry_custom branch from 8650ea6 to 320f609 Compare September 13, 2017 07:04

mariapython commented Sep 13, 2017

View reviewed changes

mariapython force-pushed the retry_custom branch from 320f609 to 3f1a714 Compare September 13, 2017 23:34

charlesccychen approved these changes Sep 14, 2017

View reviewed changes

mariapython added 4 commits September 14, 2017 18:32

Improve performance of bundle retry

8e17558

Address review comments

9f27643

Fix linter error

0c8f299

Add review comments

585d8f9

mariapython force-pushed the retry_custom branch from 3f1a714 to 585d8f9 Compare September 15, 2017 01:49

mariapython commented Sep 15, 2017

View reviewed changes

charlesccychen approved these changes Sep 15, 2017

View reviewed changes

asfgit closed this in 979923a Sep 15, 2017

Conversation

mariapython commented Sep 7, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

coveralls commented Sep 7, 2017

Uh oh!

coveralls commented Sep 7, 2017

Uh oh!

mariapython commented Sep 7, 2017

Uh oh!

charlesccychen left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mariapython left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mariapython left a comment

Choose a reason for hiding this comment

Uh oh!

coveralls commented Sep 14, 2017

Uh oh!

charlesccychen left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

coveralls commented Sep 15, 2017

Uh oh!

charlesccychen left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

mariapython commented Sep 7, 2017 •

edited

Loading