

def run(self, pipeline, node=None):
    'Remotely executes entire pipeline or parts reachable from node.'
    from google.cloud.dataflow.internal import apiclient
    self.job = apiclient.Job(pipeline.options)
    super(DataflowPipelineRunner, self).run(pipeline, node)
    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()))
        thread.daemon = True
        job_version.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/VarMisuse@165/241 `thread`->`job_version`