[BEAM-9347] Don't overwrite default runner harness for unified worker #10919

angoenka · 2020-02-21T00:04:46Z

Please add a meaningful description for your change here

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Choose reviewer(s) and mention them in a comment (R: @username).
Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

Post-Commit Tests Status (on master branch)

Lang	SDK	Apex	Dataflow	Gearpump	Samza
Go		---	---	---	---
Java
Python		---		---	---
XLang	---	---	---	---	---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website
Non-portable
Portable	---		---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

angoenka · 2020-02-21T00:06:30Z

R: @ananvay @tvalentyn

ananvay · 2020-02-21T00:30:07Z

Thanks Ankur! LGTM.

tvalentyn · 2020-02-22T01:50:28Z

sdks/python/apache_beam/runners/dataflow/internal/apiclient.py

@@ -916,6 +919,9 @@ def _use_unified_worker(pipeline_options):
  debug_options = pipeline_options.view_as(DebugOptions)
  use_unified_worker_flag = 'use_unified_worker'

+  if debug_options.lookup_experiment(use_unified_worker_flag):


PTAL at this method - we need to remove some of the statements in 922-928.

Do we need 922-924?

nvm, reading this again, looks like this logic is intenional

Ok, I think I understand the intent but I think it is confusing - can we move lines 925-926 to dataflow_runner.py, see:

beam/sdks/python/apache_beam/runners/dataflow/dataflow_runner.py

Line 480 in 1117508

# Elevate "min_cpu_platform" to pipeline option, but using the existing

, and replace 920-928 with line 928?

We can move 925-926 to dataflow_runner.py but then i feel the logic to determine unified_worker usage will be more staggered.
I think it will be better to keep the interpretation of use_unified_worker and use_runner_v2 at a single place.
For now, we support both flag but I think we will transition to use_runner_v2 over time (not sure by when).

I think that it would be cleaner to not modify experiments in _use_unified_worker, since the name of this method does not suggest that it modifies the state, and it would also be easier to follow the logic. Also, if Dataflow runner relies on the fact that "use_runner_v2" experiment is set, it might be better to set it explicitly at some point, instead of relying on the fact that "_use_unified_worker" will be called. Leaving this up to you.

tvalentyn · 2020-02-22T01:58:04Z

sdks/python/apache_beam/runners/dataflow/internal/apiclient.py

@@ -186,8 +186,11 @@ def __init__(self, packages, options, environment_version, pipeline_url):
    if job_type.startswith('FNAPI_'):
      self.debug_options = self.debug_options or DebugOptions()
      self.debug_options.experiments = self.debug_options.experiments or []
+      # Don't add the default image overwrite if user overwrites or


I think it will be easier to read if we restructure this:

if (self.debug_options.lookup_experiment('runner_harness_container_image') or _use_unified_worker(self.debug_options)): # Comment on WHY we don't want to set the override pass else: <...set the override...>

WDYT?

Also, should we be passing pipeline_options instead of debug options? Perhaps it works either way but passing pipeline options would be cleaner since that's what the signature of _use_unified_worker() expects.

I think it will be easier to read if we restructure this:

if (self.debug_options.lookup_experiment('runner_harness_container_image') or _use_unified_worker(self.debug_options)): # Comment on WHY we don't want to set the override pass else: <...set the override...>

WDYT?

I think it makes sense. Will update it.

Also, should we be passing pipeline_options instead of debug options? Perhaps it works either way but passing pipeline options would be cleaner since that's what the signature of _use_unified_worker() expects.

I think the code is structured in a way that a self.debug_options is referenced every where and the reference to it can be different from options passed in the method argument. So I think it will need a more thoughtful refactoring to move to change it to options in the right manner. Which I think we can differ.

angoenka

Thanks @tvalentyn
Made some changes and responded to some comments.

angoenka · 2020-02-24T20:19:58Z

sdks/python/apache_beam/runners/dataflow/internal/apiclient.py

@@ -916,6 +919,9 @@ def _use_unified_worker(pipeline_options):
  debug_options = pipeline_options.view_as(DebugOptions)
  use_unified_worker_flag = 'use_unified_worker'

+  if debug_options.lookup_experiment(use_unified_worker_flag):


We can move 925-926 to dataflow_runner.py but then i feel the logic to determine unified_worker usage will be more staggered.
I think it will be better to keep the interpretation of use_unified_worker and use_runner_v2 at a single place.
For now, we support both flag but I think we will transition to use_runner_v2 over time (not sure by when).

angoenka · 2020-02-24T20:28:53Z

sdks/python/apache_beam/runners/dataflow/internal/apiclient.py

@@ -186,8 +186,11 @@ def __init__(self, packages, options, environment_version, pipeline_url):
    if job_type.startswith('FNAPI_'):
      self.debug_options = self.debug_options or DebugOptions()
      self.debug_options.experiments = self.debug_options.experiments or []
+      # Don't add the default image overwrite if user overwrites or


I think it will be easier to read if we restructure this:

if (self.debug_options.lookup_experiment('runner_harness_container_image') or _use_unified_worker(self.debug_options)): # Comment on WHY we don't want to set the override pass else: <...set the override...>

WDYT?

I think it makes sense. Will update it.

Also, should we be passing pipeline_options instead of debug options? Perhaps it works either way but passing pipeline options would be cleaner since that's what the signature of _use_unified_worker() expects.

I think the code is structured in a way that a self.debug_options is referenced every where and the reference to it can be different from options passed in the method argument. So I think it will need a more thoughtful refactoring to move to change it to options in the right manner. Which I think we can differ.

tvalentyn · 2020-02-24T21:11:35Z

sdks/python/apache_beam/runners/dataflow/internal/apiclient.py

@@ -916,6 +919,9 @@ def _use_unified_worker(pipeline_options):
  debug_options = pipeline_options.view_as(DebugOptions)
  use_unified_worker_flag = 'use_unified_worker'

+  if debug_options.lookup_experiment(use_unified_worker_flag):


I think that it would be cleaner to not modify experiments in _use_unified_worker, since the name of this method does not suggest that it modifies the state, and it would also be easier to follow the logic. Also, if Dataflow runner relies on the fact that "use_runner_v2" experiment is set, it might be better to set it explicitly at some point, instead of relying on the fact that "_use_unified_worker" will be called. Leaving this up to you.

tvalentyn · 2020-02-24T21:12:28Z

sdks/python/apache_beam/runners/dataflow/internal/apiclient.py

@@ -186,8 +186,13 @@ def __init__(self, packages, options, environment_version, pipeline_url):
    if job_type.startswith('FNAPI_'):
      self.debug_options = self.debug_options or DebugOptions()
      self.debug_options.experiments = self.debug_options.experiments or []
-      if not self.debug_options.lookup_experiment(
-          'runner_harness_container_image'):
+      if self.debug_options.lookup_experiment(


Few thoughts here

I think we cannot instantiate DebugOptions() in self.debug_options = self.debug_options or DebugOptions(). We should create a view of existing options. All views of pipeline options should be views of the same underlying object, and I think that line can break this invariant if we follow the or branch. Do you know why we added it?

For the purpose of this PR it should be sufficient to pass options into _use_unified_worker() in line 190.

I agree. I think I added it because in certain case it was passed in test. I might be wrong though.

Makes sense. I will update the PR with both these suggestions.

tvalentyn · 2020-02-24T21:16:13Z

sdks/python/apache_beam/runners/dataflow/internal/apiclient.py

+      if self.debug_options.lookup_experiment(
+          'runner_harness_container_image') or _use_unified_worker(
+              self.debug_options):
+        # Don't add the default image overwrite if user overwrites or


This comment repeats the logic; I thought you might want to explain why we don't want to pin the image for UW.
Consider: "In case of using UW, harness container should be defined by DF service unless explicitly overridden."

Updates the comment.

angoenka

Thanks @tvalentyn
Updated the PR.

angoenka · 2020-02-24T21:23:50Z

sdks/python/apache_beam/runners/dataflow/internal/apiclient.py

@@ -186,8 +186,13 @@ def __init__(self, packages, options, environment_version, pipeline_url):
    if job_type.startswith('FNAPI_'):
      self.debug_options = self.debug_options or DebugOptions()
      self.debug_options.experiments = self.debug_options.experiments or []
-      if not self.debug_options.lookup_experiment(
-          'runner_harness_container_image'):
+      if self.debug_options.lookup_experiment(


I agree. I think I added it because in certain case it was passed in test. I might be wrong though.

Makes sense. I will update the PR with both these suggestions.

angoenka · 2020-02-24T21:30:10Z

sdks/python/apache_beam/runners/dataflow/internal/apiclient.py

+      if self.debug_options.lookup_experiment(
+          'runner_harness_container_image') or _use_unified_worker(
+              self.debug_options):
+        # Don't add the default image overwrite if user overwrites or


Updates the comment.

angoenka · 2020-02-24T22:35:22Z

Thanks @tvalentyn

probot-autolabeler bot added the python label Feb 21, 2020

[BEAM-9347] Don't overwrite default runner harness for unified worker

7962fa6

angoenka force-pushed the skip_harness_image_for_runner_v2 branch from 0a8d95d to 7962fa6 Compare February 21, 2020 00:38

tvalentyn reviewed Feb 22, 2020

View reviewed changes

angoenka commented Feb 24, 2020

View reviewed changes

Addressing comments.

8cd5775

tvalentyn reviewed Feb 24, 2020

View reviewed changes

angoenka commented Feb 24, 2020

View reviewed changes

Addressing more comments.

aae13bd

tvalentyn approved these changes Feb 24, 2020

View reviewed changes

angoenka merged commit e9cbb8c into apache:master Feb 24, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-9347] Don't overwrite default runner harness for unified worker #10919

[BEAM-9347] Don't overwrite default runner harness for unified worker #10919

angoenka commented Feb 21, 2020

angoenka commented Feb 21, 2020

ananvay commented Feb 21, 2020

tvalentyn Feb 22, 2020

tvalentyn Feb 22, 2020

tvalentyn Feb 22, 2020

tvalentyn Feb 22, 2020

angoenka Feb 24, 2020

tvalentyn Feb 24, 2020

tvalentyn Feb 22, 2020

tvalentyn Feb 22, 2020

angoenka Feb 24, 2020 •

edited

angoenka left a comment

angoenka Feb 24, 2020

angoenka Feb 24, 2020 •

edited

tvalentyn Feb 24, 2020

tvalentyn Feb 24, 2020 •

edited

angoenka Feb 24, 2020

tvalentyn Feb 24, 2020

angoenka Feb 24, 2020

angoenka left a comment

angoenka Feb 24, 2020

angoenka Feb 24, 2020

angoenka commented Feb 24, 2020

[BEAM-9347] Don't overwrite default runner harness for unified worker #10919

[BEAM-9347] Don't overwrite default runner harness for unified worker #10919

Conversation

angoenka commented Feb 21, 2020

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

angoenka commented Feb 21, 2020

ananvay commented Feb 21, 2020

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

angoenka Feb 24, 2020 • edited

Choose a reason for hiding this comment

angoenka left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

angoenka Feb 24, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tvalentyn Feb 24, 2020 • edited

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

angoenka left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

angoenka commented Feb 24, 2020

angoenka Feb 24, 2020 •

edited

angoenka Feb 24, 2020 •

edited

tvalentyn Feb 24, 2020 •

edited