[BEAM-759] Implement wait_until_finish method for existing runners. by aaltay · Pull Request #1762 · apache/beam

aaltay · 2017-01-10T21:49:36Z

Implement wait_until_finish method for existing runners.

Also defines the not implemented cancel() method and updates existing
usages to use wait_until_finish() instead of blocking runners.

Main changes are in the runners/ folder
runner.py - has the APIs
dataflow_runner.py, direct_runner.py modified to implement the API (moving the existing blocking code around.)

Rest of the changes are mechanical to mainly convert
p.run() to p.run().wait_until_finish() in tests and examples. Changed tests because they run validation after the run and need to block until completion. We may revert the changes in examples. I converted the because in the instructions we directed users to blocking runners before and this change keeps the behavior same.

I have started a local post commit run (not completed yet) and it was successful with the first few tests so far and the changes are same for all tests.

Remaining work after this PR:

Removing BlockingDataflowRunner. It is not used after this change with the SDK code/examples/tests.
Support for duration argument in wait_until_finish is missing.

aaltay · 2017-01-10T21:50:23Z

R: @charlesccychen

charlesccychen · 2017-01-10T22:01:15Z

sdks/python/apache_beam/examples/cookbook/datastore_wordcount.py


  # Actually run the pipeline (all operations above are deferred).
-  return p.run()
+  return p.run().wait_until_finish()


Here, we return the result. However, I don't think the result is actually used. Should we remove "return" for consistency?

result is used to get aggregated_values. I am keeping the return.

charlesccychen · 2017-01-10T22:02:09Z

sdks/python/apache_beam/internal/apiclient.py

        environment_version=self.environment_version).proto
    # TODO(silviuc): Remove the debug logging eventually.
-    logging.info('JOB: %s', job)
+    # logging.info('JOB: %s', job)


What is the intention here?

Good catch, reverted. It was my local debugging change.

aaltay · 2017-01-10T22:19:54Z

Thank you @charlesccychen.

R: @robertwb

charlesccychen · 2017-01-10T22:27:55Z

Thanks, LGTM.

asfbot · 2017-01-10T22:29:27Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/6495/
--none--

asfbot · 2017-01-10T23:13:43Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/6499/

Failed Tests: 3

beam_PreCommit_Java_MavenInstall/org.apache.beam:beam-examples-java: 2

beam_PreCommit_Java_MavenInstall/org.apache.beam:beam-runners-apex: 1

org.apache.beam.runners.apex.examples.WordCountTest.testWordCountExample

--none--

asfbot · 2017-01-10T23:47:41Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/6501/
--none--

robertwb · 2017-01-11T01:06:07Z

I think that requiring test authors to write wait_until_finished() is quite error prone. Perhaps we could make the test runner "blocking" by default (but parametrizable of course). There is no contract that p.run() return immediately, just that it can return early to provide a nicer API.

aaltay · 2017-01-11T01:11:19Z

@robertwb Thank you, I agree that, it is easy to miss wait_until_finished() in tests.

I made the TestDataflowRunner blocking, but most tests also run with DirectRunner. Are you suggesting to make DirectRunner blocking by default?

robertwb · 2017-01-11T01:29:05Z

I was suggesting making the TestRunner blocking by waiting on the underlying runner.

…

On Tue, Jan 10, 2017 at 5:11 PM, Ahmet Altay ***@***.***> wrote: @robertwb <https://github.com/robertwb> Thank you, I agree that, it is easy to miss wait_until_finished() in tests. I made the TestDataflowRunner blocking, but most tests also run with DirectRunner. Are you suggesting to make DirectRunner blocking by default? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#1762 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAdqgdfXY5kOUcn8EZGOUbzTyS8-QqMxks5rRCw5gaJpZM4Lf7vD> .

robertwb · 2017-01-13T20:31:48Z

Sorry for the confusion, I see it's a TestPipeline, not TestRunner

I think the way to do this would be to override TestPipeline.run() to return super(TestPipeline, self).run().wait_until_finished(). We could make this behavior optional, but it should default to True.

charlesccychen · 2017-01-13T20:37:14Z

Since we're on the topic, are we therefore explicitly making the decision that non-testing runners will not block on .run() by default? This makes the more common use case of .run() less intuitive.

robertwb · 2017-01-13T20:42:56Z

Runners are free to block or not as they see fit. Revisiting this for all SDKs is a bigger question.

…

On Fri, Jan 13, 2017 at 12:37 PM, Charles Chen ***@***.***> wrote: Since we're on the topic, are we therefore explicitly making the decision that non-testing runners will not block on .run() by default? This makes the more common use case of .run() less intuitive. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#1762 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAdqgU36ikJzmh1Uz4HWh6gUyhbco8Bnks5rR-B8gaJpZM4Lf7vD> .

Also defines the not implemented cancel() method and updates existing usages to use wait_until_finish() instead of blocking runners.

aaltay · 2017-01-13T22:25:19Z

@robertwb PTAL.

I updated TestPipeline and tests using that. I had to rebase because of the other changes.

asfbot · 2017-01-13T23:04:10Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/6554/
--none--

robertwb

Mostly looks good. However, I think most of the test should be converted to use TestPipeline rather than manually waiting.

robertwb · 2017-01-14T00:01:49Z

sdks/python/apache_beam/examples/complete/autocomplete_test.py

            ('that', ((1, 'that'), )),
        ]))
-    p.run()
+    p.run().wait_until_finish()


Should this be updated to use the TestPipeline instead?

Updated this and most other tests to use TestPipeline.

robertwb · 2017-01-14T00:02:41Z

sdks/python/apache_beam/examples/complete/autocomplete.py

       lambda (prefix, candidates): '%s: %s' % (prefix, candidates))
   | 'write' >> WriteToText(known_args.output))
-  p.run()
+  p.run().wait_until_finish()


Do we still need this? What happens if we don't have it? (I'd assume these are non-daemon threads we're starting, so the pipeline will still continue to run until complete before the process exits, right?

We do not need it, removed it.

DirectRunner uses non-daemon threads, and you are right pipeline continue to run until complete before the process exits.

DataflowRunner uses daemon threads. I kept the existing behavior and there is a comment about why it is needed. Pipelines still continue to finish tough because exiting from the runner after job submissions does not cancel the job.

robertwb · 2017-01-14T00:03:04Z

sdks/python/apache_beam/examples/complete/estimate_pi.py


  # Actually run the pipeline (all operations above are deferred).
-  p.run()
+  p.run().wait_until_finish()


robertwb · 2017-01-14T00:03:08Z

sdks/python/apache_beam/examples/complete/estimate_pi_test.py

    # that is very small (VERY) given that we run at least 10 million trials.
    assert_that(result, in_between(3.13, 3.15))
-    p.run()
+    p.run().wait_until_finish()


same (and the rest of the examples below). Perhaps one example should demonstrate waiting...

robertwb · 2017-01-14T00:06:28Z

sdks/python/apache_beam/examples/snippets/snippets.py

   | beam.io.WriteToText(my_options.output))

-  p.run()
+  result = p.run()


Seems a lot of these snippets could be updated to use TestPipeline, right? (At least as long as it's part of the setup/teardown code, not the actual snippet.) They're never run not under a test...

Updated most of them. I could not easily update some of these in which the pipeline creation was part of the snippet and it was immediately followed by applying transforms inside the snippet.

Looks good. I wonder if these should be part of the snippets (including constructing an empty PipelineOptions()) but we should address that as part of the documentation revamp.

Ack. Maybe. At least some of them could be part of the snippets.

robertwb · 2017-01-14T00:07:05Z

sdks/python/apache_beam/examples/snippets/snippets_test.py

    beam.assert_that(small_but_nontrivial, beam.equal_to(['bb']),
                     label='small_but_not_trivial')
-    p.run()
+    p.run().wait_until_finish()


Same for these, we should be using TestPipeline.

robertwb · 2017-01-14T00:08:13Z

sdks/python/apache_beam/io/concat_source_test.py

    assert_that(pcoll, equal_to(range(1000)))

-    pipeline.run()
+    pipeline.run().wait_until_finish()


Any reason these io tests are not using TestPipeline as well?

robertwb · 2017-01-14T00:10:24Z

sdks/python/apache_beam/runners/dataflow_runner.py


+  def _is_in_terminal_state(self):
+    if not self.has_job:
+      return True


UNKNOWN is terminal? What if self.job gets assigned later? Or can that not happen?

I am sorry I missed this in my previous reply.

It cannot happen. apiclient.DataflowApplicationClient.create_job either returns None (in case of error and template jobs) or returns a Job with a job id. It is not assigned later.

robertwb · 2017-01-14T00:19:04Z

sdks/python/apache_beam/runners/test/test_dataflow_runner.py

@@ -25,14 +25,16 @@
 class TestDataflowRunner(DataflowRunner):


All of this logic feels like it belongs in TestPipeline, not a subclass of DataflowRunner. @markflyhigh

Removed wait_until_finish except for a few exceptions: - tfidf: as an example usage. - some examples in cookbook - they run examples directly and, did not want to update the examples to use TestPipeline. - some snippets - if the pipeline creations is part of the snippet and it was not easy to override.

asfbot · 2017-01-15T08:27:48Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/6560/
--none--

aaltay · 2017-01-15T08:52:41Z

Thank you @robertwb. I updated most of the existing tests to use TestPipeline and remove wait_until_finish() from those. Except for a few places I could not easily clean. PTAL.

asfbot · 2017-01-15T09:11:00Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/6561/
--none--

robertwb

Thanks. Just a few minor comments.

robertwb · 2017-01-17T22:55:43Z

sdks/python/apache_beam/examples/snippets/snippets.py

-  p.run()
+  result = p.run()
  # [END pipelines_constructing_running]
+  result


Is this just to silence lint? Could that be done more explicitly?

No, leftover code. Removed it. (Did a a global search to find other left over code similar to this but could not.)

robertwb · 2017-01-17T22:58:19Z

sdks/python/apache_beam/examples/snippets/snippets.py

   | beam.io.WriteToText(my_options.output))

-  p.run()
+  result = p.run()


Looks good. I wonder if these should be part of the snippets (including constructing an empty PipelineOptions()) but we should address that as part of the documentation revamp.

robertwb · 2017-01-17T22:59:11Z

sdks/python/apache_beam/runners/dataflow_runner.py


+  def _is_in_terminal_state(self):
+    if not self.has_job:
+      return True


robertwb · 2017-01-17T23:00:15Z

sdks/python/tox.ini


 [testenv:py27]
 deps=
+  nose


Is this new?

New. Running autocomplete_test outside the testing framework depends on this. (I think the other option is to add it as an install_requires package since tox does setup.py install at first. Adding it as tests_require was not enough.)

aaltay · 2017-01-17T23:48:22Z

Thank you @robertwb. PTAL.

robertwb

LGTM, thanks.

asfbot · 2017-01-18T00:20:03Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/beam_PreCommit_Java_MavenInstall/6601/
--none--

aaltay · 2017-01-18T18:47:27Z

Thank you.

aaltay changed the title ~~Implement wait_until_finish method for existing runners.~~ [BEAM-759] Implement wait_until_finish method for existing runners. Jan 10, 2017

charlesccychen reviewed Jan 10, 2017

View reviewed changes

aaltay closed this Jan 10, 2017

aaltay reopened this Jan 10, 2017

aaltay added 3 commits January 13, 2017 14:00

Implement wait_until_finish method for existing runners.

37cd152

Also defines the not implemented cancel() method and updates existing usages to use wait_until_finish() instead of blocking runners.

Fix wrong use of wait_until_complete in examples.

8552f2e

Make TestPipeline.run() blocking by default.

4e0d507

aaltay force-pushed the expand branch from a277730 to 4e0d507 Compare January 13, 2017 22:21

robertwb reviewed Jan 14, 2017

View reviewed changes

Update tests to use TestPipeline()

6da0233

robertwb reviewed Jan 17, 2017

View reviewed changes

Add dependency comments to tox file.

ee1a76f

robertwb approved these changes Jan 17, 2017

View reviewed changes

asfgit pushed a commit that referenced this pull request Jan 18, 2017

Closes #1762

36a7d34

aaltay closed this Jan 18, 2017

		@@ -25,14 +25,16 @@
		class TestDataflowRunner(DataflowRunner):

Conversation

aaltay commented Jan 10, 2017

Uh oh!

aaltay commented Jan 10, 2017

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!

aaltay commented Jan 10, 2017

Uh oh!

charlesccychen commented Jan 10, 2017

Uh oh!

asfbot commented Jan 10, 2017

Uh oh!

asfbot commented Jan 10, 2017

Failed Tests: 3

beam_PreCommit_Java_MavenInstall/org.apache.beam:beam-examples-java: 2

beam_PreCommit_Java_MavenInstall/org.apache.beam:beam-runners-apex: 1

Uh oh!

asfbot commented Jan 10, 2017

Uh oh!

robertwb commented Jan 11, 2017

Uh oh!

aaltay commented Jan 11, 2017

Uh oh!

robertwb commented Jan 11, 2017 via email

Uh oh!

robertwb commented Jan 13, 2017

Uh oh!

charlesccychen commented Jan 13, 2017

Uh oh!

robertwb commented Jan 13, 2017 via email

Uh oh!

aaltay commented Jan 13, 2017

Uh oh!

asfbot commented Jan 13, 2017

Uh oh!

robertwb 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!

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!

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!

asfbot commented Jan 15, 2017

Uh oh!

aaltay commented Jan 15, 2017

aaltay Jan 17, 2017 •

edited

Loading