beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rober...@apache.org
Subject [2/3] incubator-beam git commit: Cleanup known runners code.
Date Thu, 14 Jul 2016 21:16:33 GMT
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 <robertwb@google.com>
Authored: Thu Jul 14 11:56:20 2016 -0700
Committer: Robert Bradshaw <robertwb@google.com>
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):


Mime
View raw message