def run(self, pipeline, node=None):
    """Remotely executes entire pipeline or parts reachable from node."""
    # Import here to avoid adding the dependency for local running scenarios.
    # pylint: disable=g-import-not-at-top
    from google.cloud.dataflow.internal import apiclient
    self.job = apiclient.Job(pipeline.options)
    # The superclass's run will trigger a traversal of all reachable nodes
    # starting from the "node" argument (or entire graph if node is None).
    super(DataflowPipelineRunner, self).run(pipeline, node)
    # Get a Dataflow API client and submit the job.
    standard_options = pipeline.options.view_as(StandardOptions)
    if standard_options.streaming:
      job_version = DataflowPipelineRunner.STREAMING_ENVIRONMENT_MAJOR_VERSION
    else:
      job_version = DataflowPipelineRunner.BATCH_ENVIRONMENT_MAJOR_VERSION
    self.dataflow_client = apiclient.DataflowApplicationClient(
        pipeline.options, job_version)
    self.result = DataflowPipelineResult(
        self.dataflow_client.create_job(self.job))

    if self.blocking:
      thread = threading.Thread(
          target=DataflowPipelineRunner.poll_for_job_completion,
          args=(self, self.result.job_id()))
      # Mark the thread as a daemon thread so a keyboard interrupt on the main
      # thread will terminate everything. This is also the reason we will not
      # use thread.join() to wait for the polling thread.
      thread.daemon = True
      thread.start()
      while thread.isAlive():
        time.sleep(5.0)
      if self.result.current_state() != PipelineState.DONE:
        raise DataflowRuntimeException(
            'Dataflow pipeline failed:\n%s'
            % getattr(self, 'last_error_msg', None), self.result)
    return self.result
dataset/ETHPy150Open GoogleCloudPlatform/DataflowPythonSDK/google/cloud/dataflow/runners/dataflow_runner.py/DataflowPipelineRunner.run/original