Cleanup known runners code.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/c055e845 Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/c055e845 Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/c055e845 Branch: refs/heads/python-sdk Commit: c055e845b11e8a89e26f968c11f7cb8f3943f0d5 Parents: 84fe895 Author: Robert Bradshaw <[email protected]> Authored: Thu Jul 14 11:56:20 2016 -0700 Committer: Robert Bradshaw <[email protected]> Committed: Thu Jul 14 14:16:08 2016 -0700 ---------------------------------------------------------------------- .../apache_beam/runners/dataflow_runner.py | 4 +++ sdks/python/apache_beam/runners/runner.py | 27 ++++++++++---------- 2 files changed, 18 insertions(+), 13 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c055e845/sdks/python/apache_beam/runners/dataflow_runner.py ---------------------------------------------------------------------- diff --git a/sdks/python/apache_beam/runners/dataflow_runner.py b/sdks/python/apache_beam/runners/dataflow_runner.py index 5a3f6a5..45bfb6e 100644 --- a/sdks/python/apache_beam/runners/dataflow_runner.py +++ b/sdks/python/apache_beam/runners/dataflow_runner.py @@ -44,6 +44,10 @@ from apache_beam.utils.options import StandardOptions from apache_beam.internal.clients import dataflow as dataflow_api +def BlockingDataflowPipelineRunner(*args, **kwargs): + return DataflowPipelineRunner(*args, blocking=True, **kwargs) + + class DataflowPipelineRunner(PipelineRunner): """A runner that creates job graphs and submits them for remote execution. http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c055e845/sdks/python/apache_beam/runners/runner.py ---------------------------------------------------------------------- diff --git a/sdks/python/apache_beam/runners/runner.py b/sdks/python/apache_beam/runners/runner.py index 98f9758..3a77766 100644 --- a/sdks/python/apache_beam/runners/runner.py +++ b/sdks/python/apache_beam/runners/runner.py @@ -26,6 +26,12 @@ import shutil import tempfile +_KNOWN_DIRECT_RUNNERS = ('DirectPipelineRunner', 'DiskCachedPipelineRunner', + 'EagerPipelineRunner') +_KNOWN_DATAFLOW_RUNNERS = ('DataflowPipelineRunner', + 'BlockingDataflowPipelineRunner') + + def create_runner(runner_name): """Creates a runner instance from a runner class name. @@ -40,25 +46,20 @@ def create_runner(runner_name): Raises: RuntimeError: if an invalid runner name is used. """ - # pylint: disable=wrong-import-order, wrong-import-position - if runner_name in ('DirectPipelineRunner', 'DiskCachedPipelineRunner', - 'EagerPipelineRunner'): + if runner_name in _KNOWN_DIRECT_RUNNERS: runner_name = 'apache_beam.runners.direct_runner.' + runner_name + elif runner_name in _KNOWN_DATAFLOW_RUNNERS: + runner_name = 'apache_beam.runners.dataflow_runner.' + runner_name - if runner_name in ('DataflowPipelineRunner', - 'BlockingDataflowPipelineRunner'): - import apache_beam.runners.dataflow_runner - return apache_beam.runners.dataflow_runner.DataflowPipelineRunner( - blocking=runner_name == 'BlockingDataflowPipelineRunner') - elif '.' in runner_name: + if '.' in runner_name: module, runner = runner_name.rsplit('.', 1) return getattr(__import__(module, {}, {}, [runner], -1), runner)() else: raise ValueError( - 'Unexpected pipeline runner: %s. Valid values are ' - 'DirectPipelineRunner, DataflowPipelineRunner, EagerPipelineRunner, ' - 'BlockingDataflowPipelineRunner or the fully qualified name of ' - 'a PipelineRunner subclass.' % runner_name) + 'Unexpected pipeline runner: %s. Valid values are %s ' + 'or the fully qualified name of a PipelineRunner subclass.' % ( + runner_name, + ', '.join(_KNOWN_DIRECT_RUNNERS + _KNOWN_DATAFLOW_RUNNERS))) class PipelineRunner(object):
