Implement EagerPipelineRunner, useful for running in a repl.

Using EagerPipelineRunner causes PTransforms to be evaluated eagerly
rather than being deferred.

----Release Notes----
[]
-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=122183026

Author: robertwb

google/cloud/dataflow/pipeline_test.py

@@ -244,6 +244,10 @@ def create_dupes(o, _):
             ('oom:combine/GroupByKey/group_by_window', None): 1,
             ('oom:combine/Combine/ParDo(CombineValuesDoFn)', None): 1})
 
+  def test_eager_pipeline(self):
+    p = Pipeline('EagerPipelineRunner')
+    self.assertEqual([1, 4, 9], p | Create([1, 2, 3]) | Map(lambda x: x*x))
+
 
 class Bacon(PipelineOptions):
 

google/cloud/dataflow/runners/direct_runner.py

@@ -259,3 +259,17 @@ def __init__(self, state, counter_factory=None):
 
   def aggregated_values(self, aggregator_or_name):
     return self._counter_factory.get_aggregator_values(aggregator_or_name)
+
+
+class EagerPipelineRunner(DirectPipelineRunner):
+
+  is_eager = True
+
+  def __init__(self):
+    super(EagerPipelineRunner, self).__init__()
+    self._seen_transforms = set()
+
+  def run_transform(self, transform):
+    if transform not in self._seen_transforms:
+      self._seen_transforms.add(transform)
+      super(EagerPipelineRunner, self).run_transform(transform)

google/cloud/dataflow/runners/runner.py

@@ -37,6 +37,9 @@ def create_runner(runner_name):
   if runner_name == 'DirectPipelineRunner':
     import google.cloud.dataflow.runners.direct_runner
     return google.cloud.dataflow.runners.direct_runner.DirectPipelineRunner()
+  if runner_name == 'EagerPipelineRunner':
+    import google.cloud.dataflow.runners.direct_runner
+    return google.cloud.dataflow.runners.direct_runner.EagerPipelineRunner()
   elif runner_name in ('DataflowPipelineRunner',
                        'BlockingDataflowPipelineRunner'):
     import google.cloud.dataflow.runners.dataflow_runner
@@ -45,7 +48,7 @@ def create_runner(runner_name):
   else:
     raise RuntimeError(
         'Unexpected pipeline runner: %s. Valid values are '
-        'DirectPipelineRunner, DataflowPipelineRunner, or '
+        'DirectPipelineRunner, DataflowPipelineRunner, EagerPipelineRunner, or '
         'BlockingDataflowPipelineRunner.' % runner_name)
 
 

google/cloud/dataflow/transforms/ptransform.py

@@ -387,7 +387,6 @@ def __ror__(self, left):
       p = pipeline.Pipeline(
           'DirectPipelineRunner', PipelineOptions(sys.argv))
     else:
-      deferred = True
       if not pipelines:
         if self.pipeline is not None:
           p = self.pipeline
@@ -400,6 +399,7 @@ def __ror__(self, left):
           if p != pp:
             raise ValueError(
                 'Mixing value from different pipelines not allowed.')
+      deferred = not getattr(p.runner, 'is_eager', False)
     # pylint: disable=g-import-not-at-top
     from google.cloud.dataflow.transforms.core import Create
     # pylint: enable=g-import-not-at-top