airflow/contrib/operators/dataflow_operator.py
def execute(self, context):
        """Execute the python dataflow job."""
        bucket_helper = GoogleCloudBucketHelper(
            self.gcp_conn_id, self.delegate_to)
        self.py_file = bucket_helper.google_cloud_to_local(self.py_file)
        hook = DataFlowHook(gcp_conn_id=self.gcp_conn_id,
                            delegate_to=self.delegate_to,
                            poll_sleep=self.poll_sleep)
        dataflow_options = self.dataflow_default_options.copy()
        dataflow_options.update(self.options)
        # Convert argument names from lowerCamelCase to snake case.
        camel_to_snake = lambda name: re.sub(
            r'[A-Z]', lambda x: '_' + x.group(0).lower(), name)
        formatted_options = {camel_to_snake(key): dataflow_options[key]
                             for key in dataflow_options}
        hook.start_python_dataflow(
            self.job_name, formatted_options,
            self.py_file, self.py_options)
def execute(self, context):
        bucket_helper = GoogleCloudBucketHelper(
            self.gcp_conn_id, self.delegate_to)
        self.py_file = bucket_helper.google_cloud_to_local(self.py_file)
        hook = DataFlowHook(gcp_conn_id=self.gcp_conn_id,
                            delegate_to=self.delegate_to,
                            poll_sleep=self.poll_sleep)
        dataflow_options = self.dataflow_default_options.copy()
        dataflow_options.update(self.options)
        # Convert argument names from lowerCamelCase to snake case.
        camel_to_snake = lambda name: re.sub(
            r'[A-Z]', lambda x: '_' + x.group(0).lower(), name)
        formatted_options = {camel_to_snake(key): dataflow_options[key]
                             for key in dataflow_options}
        hook.start_python_dataflow(
            self.job_name, formatted_options,
            self.py_file, self.py_options)
def execute ( self , context ) : bucket_helper = GoogleCloudBucketHelper ( self . gcp_conn_id , self . delegate_to ) self . py_file = bucket_helper . google_cloud_to_local ( self . py_file ) hook = DataFlowHook ( gcp_conn_id = self . gcp_conn_id , delegate_to = self . delegate_to , poll_sleep = self . poll_sleep ) dataflow_options = self . dataflow_default_options . copy ( ) dataflow_options . update ( self . options ) # Convert argument names from lowerCamelCase to snake case. camel_to_snake = lambda name : re . sub ( r'[A-Z]' , lambda x : '_' + x . group ( 0 ) . lower ( ) , name ) formatted_options = { camel_to_snake ( key ) : dataflow_options [ key ] for key in dataflow_options } hook . start_python_dataflow ( self . job_name , formatted_options , self . py_file , self . py_options )
Execute the python dataflow job.
Execute the python dataflow job .