'Execute the python dataflow job.'

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)
    dataflow_options = self.dataflow_default_options.copy()
    dataflow_options.update(self.options)
    camel_to_snake = (lambda name: re.sub('[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.task_id, formatted_options, self.py_file, self.py_options)
